From ec82d95a03316cd82f83bac78ccd0dbaec13cc38 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 10 Oct 2026 23:13:24 +0200 Subject: [PATCH] Add signed FEC stream core and source buffering --- doc/stream.txt | 517 +++++++++++++++++++++++++++++++++++- doc/tasks.md | 8 + lib/debug_config.c | 1 + lib/debug_config.h | 3 +- src/Makefile.am | 13 + src/stream/stream_block.c | 180 +++++++++++++ src/stream/stream_block.h | 48 ++++ src/stream/stream_buffer.c | 330 +++++++++++++++++++++++ src/stream/stream_buffer.h | 53 ++++ src/stream/stream_metrics.c | 136 ++++++++++ src/stream/stream_metrics.h | 64 +++++ src/stream/stream_proto.h | 73 +++++ src/stream/stream_source.c | 181 +++++++++++++ src/stream/stream_source.h | 36 +++ tests/Makefile.am | 5 + tests/test_stream_core.c | 326 +++++++++++++++++++++++ 16 files changed, 1960 insertions(+), 14 deletions(-) create mode 100644 src/stream/stream_block.c create mode 100644 src/stream/stream_block.h create mode 100644 src/stream/stream_buffer.c create mode 100644 src/stream/stream_buffer.h create mode 100644 src/stream/stream_metrics.c create mode 100644 src/stream/stream_metrics.h create mode 100644 src/stream/stream_proto.h create mode 100644 src/stream/stream_source.c create mode 100644 src/stream/stream_source.h create mode 100644 tests/test_stream_core.c diff --git a/doc/stream.txt b/doc/stream.txt index 3705230a..106eae36 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -1,20 +1,511 @@ -Модуль стриминга. Архитектурные нюансы. -главная цель: узлы ищут более оптимальные пути доставки контента от источника до себя. +Модуль стриминга. Техническое задание, этап 1. -этап 1. -- реализуем в utun отдельным каталогом: src/stream -- источников стриминга может быть несколько в группе. для каждого источника в группе создается свой экземпляр. В этом экземпляре: - - отдельные списки хендлов подключений (ncd) для стриминга. - - узлы откуда берем трафик - - узлы куда отдаём трафик +1. Цель и принятые решения +Модуль размещается в src/stream. Узлы получают поток данных от источника через других подписчиков группы, +передают его своим подписчикам и постоянно ищут более качественные пути доставки отдельных субпотоков. +Этап 1 использует CHAT-группы с таблицей мемберов и member_sync/merkle_sync. -формат передачи: - Источник формирует несколько потоков с разным битрейтом (до 3х штук). +Принятые решения: +- Транспорт — существующий ETCP с быстрыми переповторами, мультилинком, шифрованием и управлением нагрузкой. +- Параметры FEC задаёт источник; 6+2 — значение по умолчанию, а не ограничение протокола. +- Источник подписывает каждый блок после разбиения на субпотоки, включая проверочные блоки. +- В источник добавляется циклический timestamp uint16_t в миллисекундах; он сохраняется при всей ретрансляции. + Задержка оценивается относительно предыдущих данных по дельтам timestamp и времени прихода, включая полную остановку. +- Передаём произвольные данные как есть, без обработки форматов видео, кадров и ключевых точек. +- Размер одной кодограммы после разбиения — около 1300 байт; источник сбрасывает неполную пачку через 100 мс. +- На каждый локальный экземпляр трансляции — кольцевой буфер 32 МБ для всех её субпотоков вместе. + Вся выдача идёт из этого буфера. Вытесненные данные недоступны: не успел получить — опоздал. +- Отдельных дозапросов диапазонов нет. Единственный способ получения — запросить субпоток с индекса или с конца. +- Обратное подключение — NCD + REVERSE_REQ стриминга через маршрутизатор группы. +- Статус источника и список подписок — опции самого мембера, распространяемые через Merkle. + Удалённая подписка актуальна только пока узел онлайн в BGP данной группы. +- Экземпляры определяются локальными подписками; узлы без подписки не принимают и не ретранслируют поток. + Активный источник также имеет собственный экземпляр и подписку на свою трансляцию. +- Подтверждение перегрузки — 500 мс. Число отключаемых выдач определяется ростом очереди и rate отдельных выдач. +- При запуске параллельно пробуем до 32 совместимых кандидатов; далее одна проба в секунду, + новый полный цикл не начинается раньше чем через 30 секунд от начала предыдущего. +Обратная совместимость с прежними форматами не требуется. +2. Трансляция, подписки и владение ресурсами -Надо учитывать что узлы могут быть перегружены трафиком. 2 сигнала перегрузки: очередь отправки и поток uasync Не справляется с нагрузкой (uasync thread load). +Трансляция определяется ключом (group_id, source_node_id, source_epoch). +source_epoch — новый случайный ненулевой 64-битный идентификатор при каждом запуске источника. +В этапе 1 у одного источника в одной группе активна одна трансляция. Старые и новые запуски различаются по эпохе. +Пакеты и уведомления старого запуска не могут менять состояние нового. -- узлы учавствующие в стриминге анонсируют себя аналогично как сделано в рации. -У модуля свой фоновый опросчик связности и поиска хороших линков - он пробует связь с узлами стриминна кроме мобильных, и обновляет их rtt и флаг прямой доступности. +Локальная подписка на трансляцию создаёт один экземпляр, общий для получения и ретрансляции. +Подписка на трансляцию отличается от сетевой выдачи конкретного субпотока по конкретному handle. +Один экземпляр может получать субпотоки от нескольких пиров и выдавать их нескольким получателям. +Повторная локальная подписка не создаёт второй буфер; разные локальные потребители используют один экземпляр. +Закрытие последнего локального владельца подписки останавливает экземпляр и все его сетевые выдачи. + +Экземпляр хранит: +- Дескриптор источника, FEC-контекст, состояние и номер эпохи. +- Собственные NCD handles, независимо от владельцев транспорта в других модулях и трансляциях. +- Upstream-выдачи: поставщик, handle, субпоток, текущая позиция, путь, прогресс и метрики. +- Downstream-выдачи: получатель, handle, субпоток, курсор в кольце и срок аренды. +- Кольцевой буфер, индекс наличия блоков, состояния сборки пачек и статистику. +- Лимит числа одновременно выдаваемых субпотоков. + +Одна выдача — один субпоток одному получателю. Один субпоток двум получателям занимает два исходящих слота. +Есть также общий лимит исходящих выдач модуля: лимиты разных экземпляров не должны суммироваться без ограничения. +Модуль хранит общий контекст пира в пределах UTUN_INSTANCE: одну очередь отправки стриминга для всех его +трансляций и групп, waiter на транспорт, stream-сессию, метрики и текущий запрос разгрузки. + +Увиденная удалённая опция мембера не создаёт локальную подписку, буфер или NCD-подключение. +Без локального экземпляра dispatcher отбрасывает DATA после минимального разбора ключа трансляции: +не проверяет подпись содержимого, не выполняет FEC, не буферизует и не пересылает блок. +INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получает явный отказ, а не создаёт неявную подписку. + +Операции с экземплярами, NCD, очередями, Merkle и таймерами выполняются в uasync-потоке ядра. +Внешние producers передают байты через существующий механизм post с ограничением ожидающего объёма. +При нехватке ресурсов создание подписки завершается явной ошибкой с очисткой частично созданного состояния. + +3. FEC, номера блоков и размер кодограммы + +Источник формирует пачки и кодирует их lib/reed_solomon: +- n — полное число субпотоков, r — число проверочных, k = n-r — число исходных. +- По умолчанию n=8, r=2, k=6. Любые k кодограмм одной пачки восстанавливают её содержимое. +- subblock от 0 до n-1 определяет субпоток; первые k блоков систематические, остальные проверочные. +- batch_seq — общий 64-битный индекс пачки; блок субпотока X определяется парой (batch_seq, X). + Позиция N в запросе — batch_seq, а не байтовое смещение или отдельный счётчик субпотока. +- Номера возрастают без повторов; UINT64_MAX зарезервирован для запроса с конца. + Переполнение требует новой source_epoch. + +Дескриптор содержит ключ трансляции, n, r и размер RS-кодограммы block_size. +Параметры задаёт только источник и не меняет в пределах source_epoch; изменение создаёт новый запуск. +Дескриптор подтверждается подписанной опцией источника; поля FEC также входят в подписанный блок. +Ограничения кодека: 1 <= n <= 255, 0 <= r < n, 10 <= block_size <= 4096. +Кроме допустимости RS проверяется расход памяти и корректность полного размера блока. + +Целевой размер неизменяемого блока субпотока — около 1300 байт, включая метаданные источника и его 64-байтную подпись. +block_size для RS рассчитывается из этого бюджета; он включает 9-байтный заголовок RS seq/subblock. +Handle получателя, поколение stream-сессии, изменяемый путь и служебные заголовки ETCP идут сверх этого бюджета. +Не обещаем соответствия одной кодограммы одному UDP-пакету: необходимую фрагментацию выполняет ETCP normalizer. + +На первой порции данных неполной пачки источник запускает таймер 100 мс. +При заполнении пачка немедленно кодируется, таймер отменяется; для следующей неполной пачки начинается новый срок. +По таймауту кодируем накопленные данные с дополнением нулями, сохраняя их действительную длину. +Без накопленных данных пустые пачки не создаём. При штатном окончании источник сразу сбрасывает остаток. +Этот таймер есть только у источника; ретрансляторы не накапливают данные для повторного формирования пачек. + +В этапе 1 ретранслятор пересылает полученные блоки; отсутствующие блоки для downstream заново не генерирует. +Локальный decode восстанавливает содержимое для потребителя, но не создаёт подписи источника на отсутствующие блоки. +Восстанавливаем при наличии k проверенных блоков, не ожидая остальных. +Полученные байты выдаются как есть с удалением padding, по порядку batch_seq. +Невосстановимый пропуск обозначается событием GAP; нельзя молча склеивать данные по обе стороны разрыва или выдавать нули за исходные байты. +Окончательно пропускаем пачку, когда её данных уже нет в окне либо все нужные выдачи прошли её индекс/сообщили пропуск, +а k блоков не собрано. Получение более поздней пачки от одного поставщика само по себе не закрывает ожидание остальных. +Модуль не интерпретирует содержимое и не занимается видео, кодированием, воспроизведением или поиском ключевых кадров. + +4. Подпись источника и timestamp + +Каждый полученный после RS блок, в том числе проверочный, подписывается Ed25519-ключом самого источника. +Подписывается каноническое сообщение с маркером назначения подписи stream_block, включающее: +group_id, source_node_id, source_epoch, параметры FEC, batch_seq, subblock, source_ts_ms, +действительную длину содержимого пачки, флаги источника и все байты RS-кодограммы. +Byte order и точная раскладка фиксируются в stream_proto.h. Дублирующиеся поля seq/subblock проверяются на совпадение. +Это связывает блок с группой, запуском, индексом, временем и параметрами кодирования. + +Подпись создаётся один раз в источнике после FEC. Она хранится с блоком и пересылается без изменений. +Handle, stream-сессия и hop-list находятся вне подписанного тела, поскольку меняются при выдаче и ретрансляции. +Перед помещением нового блока в кольцо, FEC или ретрансляцией проверяется подпись по ключу источника +из проверенной записи мембера данной группы. Ключ непосредственного поставщика не заменяет ключ источника. +Неверная подпись, неизвестный ключ или конфликт метаданных — отказ с диагностикой, без изменения индекса наличия. +Для одинакового уже проверенного блока допустимо повторно не выполнять crypto, только если совпадает всё неизменяемое тело. +Одинаковый индекс с другим содержимым нельзя считать обычным дубликатом без проверки. + +source_ts_ms = (uint16_t)monotonic_time_ms источника в момент формирования пачки. +У всех её субпотоков одинаковые timestamp, длина и флаги. Ретранслятор не переписывает их своим временем. +Timestamp циклический с периодом 65536 мс; обычный переход через ноль обрабатывается модульной разностью. +Это не Unix-время; NTP и сравнение абсолютных часов разных узлов для метрики задержки не требуются. + +16 бит не позволяют однозначно восстановить произвольное число полных оборотов. +Источник помечает в подписанных флагах разрыв временной шкалы, если интервал между его пачками >= 65536 мс. +При таком флаге, смене поставщика/эпохи, потере непрерывности индексов или длинном перерыве приёма +получатель начинает новую базу измерения, не придумывая число оборотов. +Первый блок устанавливает базу и не измеряет абсолютную задержку пути. + +5. Кольцевой буфер и единственный способ выдачи + +На экземпляр — 32 * 1024 * 1024 байт под неизменяемые блоки всех субпотоков вместе, включая их подписи. +Метаданные индекса, путей и сборки учитываются отдельно и ограничиваются. +Буфер общий для локального потребителя и всех downstream; копии кольца на каждого получателя не создаются. +Источник сначала помещает сформированные блоки в своё кольцо, затем делает их доступными выдачам. +Ретранслятор сначала проверяет и сохраняет блок, затем планирует выдачу подписчикам. + +Индекс записи — (batch_seq, subblock) внутри конкретной эпохи. +Новые пачки вытесняют старейшие; при вытеснении пачки удаляются все её блоки и метаданные. +Дубликаты не занимают повторную память и не продлевают хранение. +Блоки старше нижней границы окна не вытесняют более свежие данные. +По каждому субпотоку известны доступные границы и наличие блоков внутри них; между границами возможны дырки. +Длительность истории определяется битрейтом и FEC; фиксированного времени хранения нет. + +Есть только SUBSCRIBE(subblock, N): +- Конкретный N: начать непрерывную выдачу этого субпотока с указанного индекса, затем выдавать новые блоки. +- N=UINT64_MAX: начать с последнего доступного блока запрошенного субпотока, затем продолжать выдачу. +Нет REPAIR_REQ, выдачи диапазонов и отдельной системы повторного запроса потерянных блоков. + +Выдача хранит курсор и читает только кольцо; запрос не закрепляет данные в памяти и не резервирует историю. +Если начальный блок уже вытеснен или недоступен, возвращаем явный отказ с границами окна. +Недоступную позицию не заменяем последней молча. +Если курсор работающей выдачи отстал за вытесненные данные, сообщаем EXPIRED/GAP для пропуска и продолжаем с доступного индекса. +Если внутри окна нет блока, но уже доступны более поздние, сообщаем GAP и продолжаем. +Восстановление получения выполняется выбором поставщика/новым SUBSCRIBE с нужного индекса, пока он ещё есть в его кольце. +Никакого отдельного архивирования, гарантии хранения до доставки или ожидания медленного получателя нет. + +Общая очередь хранит ключ блока и идентификатор выдачи; долговременно указатели внутрь переиспользуемого кольца не сохраняются. +Перед передачей в ETCP наличие проверяется заново, пакет получает самостоятельное содержимое. +Вытесненное задание учитывается как пропуск, отменённое не отправляется. +После CLOSED не отправляются новые DATA/GAP по старому handle. +Уже принятые транспортом пакеты принадлежат ETCP и не ссылаются на память кольца. + +6. Транспорт и NCD + reverse + +Данные и управление установленной выдачей идут непосредственно между поставщиком и получателем через raw ETCP. +Для стриминга выделяются новые IDs в едином реестре src/transport_layer/etcp_api.h: +raw ID прямого обмена и router service ID управления через группу, без конфликтов в соответствующих пространствах. +ETCP обеспечивает переповторы, порядок, мультилинк и backpressure; второй механизм транспортных ACK поверх него не создаётся. + +До отправки требуется UP от собственного NCD handle. Наличие ETCP_CONN само по себе недостаточно. +Согласование BGP-сессий остаётся в topo_group; stream приобретает своё NCD-владение и не дублирует этот протокол. +Членство проверяется отдельно; NCD такую проверку не выполняет. +На стороне отправителя до ACCEPT приобретается собственное NCD-владение на срок выдачи. + +Прямое подключение запускается через node_conn_direct_open(). +При необходимости инициатор отправляет REVERSE_REQ через маршрутизатор группы; +получатель проверяет членство, наличие своей подписки на эту трансляцию и открывает встречный NCD handle. +Запрос содержит ID попытки и ключ трансляции; адреса и ключ пира берутся из проверенного реестра. +Повтор одной попытки не создаёт дополнительные handles; сроки и повторы ограничены. +Завершение/отмена попытки убирает её таймеры и временное владение. + +Совместимость учитывает оба узла, локальные сокеты, типы NAT, LAN/публичные IPv4/IPv6 и direct/reverse. +Используются существующие критерии sock_match/NCD; отбор не строится только по наличию EIM у одного пира. +Код адресного отбора, handshake, шифрования и линков в stream не дублируется. +Маршрутизатор группы обслуживает REVERSE_REQ и уведомления при отсутствии прямой связи, +а не большую INDIRECT-выдачу через conn_mgr. + +7. Протокол и сроки выдач + +Основные команды: +- INFO_REQ/INFO_RESP — дескриптор, доступные субпотоки, диапазоны кольца, пути, rates и метрики кандидата. +- SUBSCRIBE/ACCEPT/REJECT — непрерывная выдача с индекса либо последнего доступного блока. +- RENEW — продление списка существующих выдач, период 2 секунды. +- UNSUBSCRIBE/CLOSED — закрытие выдачи с указанием handle и причины. +- DATA/GAP/EXPIRED — подписанный блок либо уведомление о пропуске. +- OVERLOAD — запрос сокращения исходящей нагрузки через закрытие нескольких выдач. +- REVERSE_REQ — встречное NCD-подключение через маршрутизатор группы. +- END — последняя пачка и штатное завершение содержимого источником. + +handle — ненулевой 64-битный ID, назначаемый получателем. +Предлагаемый простой механизм защиты от повторов: возрастающие handles в пределах stream-сессии пира, +новые SUBSCRIBE отправляются по порядку в одной управляющей FIFO. +Номер выделяется непосредственно при постановке нового SUBSCRIBE в эту FIFO, после готовности подключения; +параллельные ожидания NCD/INFO не резервируют номера, которые позже нарушили бы порядок отправки. +Ключ выдачи — peer, stream key, поколение stream-сессии и handle. +Повтор известного действующего SUBSCRIBE с прежними параметрами возвращает прежний результат, +не создаёт вторую выдачу, не меняет курсор и продлевает аренду. +Другие параметры по тому же handle отвергаются. +Для закрытого/неизвестного handle не выше уже обработанного номера SUBSCRIBE — STALE_REQUEST. +Водяной знак сохраняется до конца сессии; не нужен неограниченный список всех ранее закрытых handles. +RENEW создаёт только продление существующей выдачи. Повторное закрытие безопасно. +Для другой позиции создаётся новый handle; после переполнения счётчика требуется новое поколение сессии. + +Поколения пиров согласуются при прямом INFO-обмене до SUBSCRIBE. +При DOWN, REINIT или закрытии старое поколение инвалидируется вместе с его выдачами. +Поздние ACCEPT/DATA/routed CLOSED не восстанавливают их. ETCP_CONN может пережить REINIT, его адрес не является ID сессии. +Управляющие поля и размеры списков имеют явный byte order и пределы. + +ACCEPT содержит handle, фактический начальный индекс, границы кольца и путь субпотока. +Реальное получение блока, а не только ACCEPT, подтверждает пригодность выдачи. +Отказы различают отсутствие подписки/трансляции, эпоху, FEC, позицию, слот, перегрузку, цикл и право доступа. +DATA содержит поколение сессии, handle, неизменяемый подписанный блок источника и изменяемый путь. +Источник подписывает и END с ключом трансляции и индексом последней пачки, чтобы ретранслятор не мог подделать окончание. + +Нет продления 6 секунд — отправитель закрывает выдачу и сообщает CLOSED, при необходимости через маршрутизатор группы. +Аренда, пробы и timers используют монотонное локальное время. Таймаут аренды очищает ресурсы; +реакция на отсутствие прогресса начинается раньше по метрикам потока. +Потеря отдельного ETCP-линка не закрывает выдачи, пока NCD остаётся UP. +DOWN закрывает выдачи соответствующей сессии; REINIT требует их согласовать заново. +Локальное освобождение ресурсов не ждёт доставки уведомления. + +8. Начальная позиция и перенос субпотоков + +Начальный запуск: +1) По опциям мемберов и BGP выбрать первый подходящий онлайн-узел, включая сам источник. +2) Запросить один субпоток с N=UINT64_MAX. +3) Получить и проверить первый блок, зафиксировать его source_epoch и batch_seq=N. +4) У остальных поставщиков запросить субпотоки этой же эпохи с конкретного N. +5) Собрать достаточный набор блоков одной пачки, затем продолжать непрерывное получение. + +Не выбирать независимый последний индекс у каждого поставщика: пачки должны совпадать. +Если N успел исчезнуть из колец или набор собрать нельзя, явно начать новую попытку с конца одного поставщика. +Не смешивать старую и новую попытки и не выдавать их за непрерывное содержимое. + +Обычный перенос одного субпотока: +1) Выбрать первую ещё нужную пачку N и поставщика, способного её выдать. +2) Создать новую выдачу, продолжая старую; временное перекрытие также занимает исходящий слот нового поставщика. +3) Получить блоки и подтвердить продвижение нового поставщика. +4) Закрыть старую выдачу. При отказе новой сохранить старую и отменить неудачную попытку. + +Дедупликация — по (stream key, batch_seq, subblock), независимо от handles. +При перекрытии качество поставщиков учитывается отдельно, но блок потребителю выдаётся один раз. +При явном отказе разрешено одновременно восстанавливать несколько субпотоков. +Точная задержка обычного переноса и гистерезис задаются настройками; небольшое колебание RTT не запускает перестройку. + +9. Пути и защита от циклов + +У каждого субпотока свой stream-путь от источника до поставщика; BGP-путь его не заменяет. +INFO_RESP и ACCEPT содержат путь запрошенного субпотока. +Путь начинается источником, не содержит нулей и повторов; подписка запрещена, если получатель уже находится в нём. +Эта проверка повторяется при принятии запроса, а не только при предварительном выборе. + +DATA несёт фактически пройденные узлы; начальный путь содержит источник, последний — непосредственного проверенного отправителя. +При пересылке добавляется свой node_id. Повторное посещение и превышение максимума хопов запрещены. +При выдаче из кольца сохраняется история пути блока; запрещено отправлять его уже пройденному узлу. +В связи с этим отсутствие пригодного блока для конкретного получателя также обозначается пропуском. +Смена upstream обновляет объявляемый путь и повторно проверяет downstream-выдачи; возникший цикл закрывается с причиной LOOP. +При гонках метрик временно возможна неработающая цепочка, но DATA не циркулирует, а отсутствие прогресса запускает поиск. +Старые данные в кольце сами по себе не делают узел качественным live-поставщиком. + +10. Относительная задержка, остановка и rate + +Метрики ведутся отдельно для каждого поставщика и субпотока. +Обрабатываются новые индексы, а не дубликаты; timestamp всех субпотоков одной пачки одинаковый. +Для последовательных блоков: + source_delta_ms = (uint16_t)(source_ts_current - source_ts_previous) + receive_delta_ms = local_receive_current_ms - local_receive_previous_ms + lag_ms = max(0, lag_ms + receive_delta_ms - source_delta_ms) + +Первый блок устанавливает базу. Обычный оборот timestamp обрабатывается без скачка lag. +Так измеряется накопленное отставание относительно скорости источника, а не абсолютное время от источника до получателя. +Ускоренный приход уменьшает lag, но значение не становится отрицательным. +Источник timestamp не изменяется по пути, поэтому изменение доставки учитывает задержки на всей цепочке. +Базу сбрасываем при отмеченном источником разрыве времени, неоднозначной дельте, новой выдаче/эпохе или разрыве последовательности. +Резкое изменение позиции выдачи не считается доказательством улучшения задержки. + +Ожидаемый межблочный интервал и rate источника оцениваются по окну его timestamp и объёмам блоков. +Если несколько блоков имеют один timestamp, агрегируем объём до ненулевой дельты, не делим на ноль. +Не используем время формирования всех n блоков одной пачки как n независимых временных интервалов. +При отсутствии новых данных продолжаем измерять отставание от ожидаемого следующего прихода: + stall_lag_ms = lag_ms + max(0, now_ms - last_receive_ms - expected_interval_ms) +Превышение 500 мс является проблемой доставки; полная остановка не остаётся незамеченной только потому, что новых timestamp нет. +Отсутствие данных источника и остановка на маршруте по одной этой метрике неразличимы: +фиксируем отсутствие прогресса, не приписывая ему недоказанную причину. END прекращает ожидание новых блоков. + +История проблем — 60 секундных корзин: число запаздывающих блоков и отдельные эпизоды остановки. +Блок считается запоздавшим один раз для соответствующей выдачи; один эпизод остановки не считается заново на каждом тике. +Сохраняются число наблюдаемых блоков, rates и доля проблем, чтобы сравнение не искажалось разным битрейтом. +FEC-восстановление не считается доставкой отсутствующего субпотока; частичные пачки и пропуски считаются отдельно. +Скорость продвижения по старой истории при запуске/переносе отличается от steady live и явно отмечается. + +INFO_RESP также сообщает число свободных исходящих слотов, диапазоны буфера, актуальность прогресса, очередь и нагрузку. +Суммарный RTT = RTT непосредственного соединения + объявленный RTT stream-пути кандидата, отдельно по субпотоку. +Поле суммарного RTT — не менее 32 бит с явными единицами и признаком неизвестного/устаревшего значения. + +11. Поиск поставщиков и расписание проб + +Кандидаты — онлайн в BGP данной группы, с подпиской на нужную трансляцию/эпоху в опциях мембера. +Начальный поиск учитывает источник. Фоновые пробы ретрансляторов исключают мобильных. +Мобильный узел может быть источником и получателем; выбранный мобильный источник допускается к подключению. +Активное получение/публикация согласовывается с активностью ядра, чтобы standby не прерывал поток. + +При запуске экземпляра одновременно пробуем до 32 совместимых по типу подключения кандидатов. +Если подходящих меньше, пробуем имеющихся. Это узлы, а не 32 отдельных адреса или субпотока. +Совместимость определяется существующими транспортными правилами; известный рабочий транспорт переиспользуется. +После стартовой серии запускаем не более одной новой пробы в секунду. +Один кандидат пробуется один раз за цикл; новый полный цикл начинается не раньше 30 секунд от старта предыдущего. +Если кандидатов мало и цикл исчерпан раньше, до этого срока ждём. Если их много, цикл занимает фактическое время обхода. +Стартовые пробы учитываются в первом цикле, а не повторяются сразу после него. +Новые кандидаты из Merkle/BGP могут добавляться в текущий цикл с темпом 1/с, без повторного опроса уже проверенных. +Неуспех имеет backoff; одинаковые активные пробы одного пира для разных экземпляров переиспользуются. +Временные NCD handles невыбранных кандидатов освобождаются; рабочие экземпляры удерживают свои handles. + +Используются route_connectivity, sock_match/NCD и метрики существующих соединений; второй адресный probe-движок не создаётся. +Прямая доступность — UNKNOWN/REACHABLE/UNREACHABLE с временем замера, а не вечный boolean. + +Порядок выбора: доступный субпоток/индекс и отсутствие цикла; затем проблемы доставки и RTT; +далее устойчивость набора поставщиков, совпадение промежуточных узлов и число хопов. +Стремимся получить n субпотоков, хотя k достаточно для decode. +При возможности не более r субпотоков от одного поставщика: потеря его при получении всех n оставляет не менее k. +При r=0 такой защиты нет; при недостатке кандидатов ограничение ослабляется с отражением устойчивости в статусе. +Учитываем совпадение путей разных поставщиков, кроме общего источника. + +12. Общая отправка и перегрузка + +На удалённого пира — одна ограниченная очередь стриминга для всех групп и источников. +Управление имеет небольшой приоритет; выдачи данных чередуются, чтобы один источник не захватил всю очередь. +При старте с прошлого индекса не помещаем сразу всё кольцо в очередь: читаем его по курсору по мере готовности отправки. +Темп задают имеющиеся данные и backpressure ETCP, независимый транспортный shaper не создаётся. + +Отправка ждёт освобождения conn->send_input_q через queue_waiter_wait с существующим порогом. +Не переполняем очередь, не меняем чужой общий порог и не заменяем waiter активным опросом. +Handle waiter обнуляется полностью, callback может быть синхронным; отменяем waiter до освобождения владельца. +При автозаборе ll_queue соблюдается queue_data_get -> обработка -> queue_resume_callback. +При успехе etcp_send пакет принадлежит транспорту; при ошибке его освобождает stream. +Принятие стеком не означает доставку; для диагностики учитываются нижние очереди/inflight и продвижение приёмника. +Уже принятые ETCP пакеты не извлекаются обратно; общий транспорт ради одного потока не сбрасывается. + +Раз в 200 мс сохраняется интервал, история — 10 секунд. +Длительность интервала фактическая, включая задержки uasync. +По каждому пиру и выдаче считаем пакеты/байты: +A — поступившая нагрузка, D — передано в ETCP, X — отменено/вытеснено, Q — ожидающий объём; +баланс Q_next = Q + A - D - X, возраст старейшего задания, rate и скорость роста. +Для заданий-ключей байтовый объём — размер ожидающего блока, память самих заданий учитывается отдельно. +Также сохраняются отказы при постановке, чтобы ограничение очереди не скрывало реальный избыток нагрузки. + +Подтверждение перегрузки — непрерывные 500 мс наблюдаемого отставания, по монотонному времени, не по округлённому числу тиков. +Возраст данных >200 мс вместе с устойчивым ростом — сигнал недостаточной скорости. +Полная остановка D также обрабатывается. Краткий глитч с последующей быстрой разгрузкой не вызывает немедленного отключения. +Нагрузка uasync берётся из uasync_get_load(); дополнительно измеряется опоздание callbacks. +Проблема одного пира отличается от одновременного роста многих очередей и локального отставания event loop. +Высокий CPU отдельно от отставания/очередей не является достаточным основанием. + +Скорость роста g оценивается в байтах/с по увеличению Q до искусственных удалений: +в интервале без смены выдач g = max(0, (A-D)/elapsed_seconds). +Rate выдачи — её средний вклад в поступающую нагрузку, а не урезанная скорость выхода из перегруженного канала. +Для разгрузки нужно снять суммарный rate больше g, чтобы очередь начала убывать. +Число выдач выбирается по их rates: закрываем достаточно выдач для требуемого сокращения; +при сильном росте это сразу несколько, не фиксированная одна и не случайное постоянное число. + +OVERLOAD содержит request_id, расчётное число выдач и требуемое сокращение rate. +Получатель выбирает handles с учётом FEC и альтернативных поставщиков. +При разных rates одного количества закрытий недостаточно: суммарный снятый rate должен покрыть требование. +Rate закрываемых выдач фиксируется по оценке на момент запроса, чтобы условие исполнения не менялось на каждом тике. +Срок реакции — 2 секунды с первого запроса; повторы не сбрасывают срок и не начинают новое требование. +Новые выдачи, компенсирующие запрошенное сокращение, до разгрузки не принимаются. +Если требование не выполнено, отправитель сам закрывает достаточный остаток выдач и сообщает handles/причину. +Далее снова измеряется рост; автоматический выбор учитывает освобождаемую нагрузку, а при равенстве допустима случайность. +При глобальном отставании сокращение выбирается по всем пирам модуля. +Память и возраст очереди остаются ограниченными во время ожидания; просроченные задания удаляются с GAP/EXPIRED. + +13. Статусы через опции мембера и Merkle + +Статус источника и список подписок принадлежат самому мемберу в конкретной группе. +Они входят в его подписанный блок A/userinfo и распространяются существующим member_sync/merkle_sync. +adm_tags — блок владельца, для самоуправляемых stream-опций он не используется. +Отдельного STREAM_ANNOUNCE, broadcast-обновлений и собственных TTL/версий анонсов нет. + +Опции содержат: +- stream_source: активна ли собственная трансляция; если активна — её source_epoch и параметры FEC. +- stream_subscriptions: список (source_node_id, source_epoch), на которые узел локально подписан. +Формат — совместимые с существующим плоским userinfo строковые значения; список кодируется однозначно, +со строгим разбором 64-битных IDs без потери точности через JSON numbers. +Источник включает свою трансляцию в собственные подписки. +Список относится к трансляциям, а не к сетевым handles или выбранным upstream-поставщикам. + +Изменение подписки/запуск/остановка источника обновляет свой блок мембера: +userinfo + возрастающий update_ts + update_sig, атомарно с Merkle-хешами, после commit запускается распространение. +Другие опции (имя, аватар и т.д.) сохраняются; запись одного модуля не должна стирать состояние другого. +Размер укладывается в MS_USERINFO_MAX; переполнение — явная ошибка подписки, а не молчаливое обрезание. +При запуске ядра свои stream-опции приводятся к реально восстановленному локальному состоянию, +чтобы старый источник/подписка после аварии не объявлялись работающими без экземпляра. + +Подписка удалённого мембера актуальна только если есть его живой BGP-путь в этой же группе. +Недостаточно присутствия записи в SQLite/реестре или произвольного ETCP-соединения из другой группы. +При исчезновении BGP-пути узел исключается из кандидатов; его подписанная опция в Merkle не удаляется третьей стороной. +При возвращении в BGP опции читаются заново и текущая доступность подтверждается INFO. +Статус источника также проверяется совместно с его доступностью; наличие старого source_epoch в БД не доказывает поступление данных. + +Stream следит за двумя независимыми событиями: изменением блока мембера и изменением BGP-доступности. +Текущий props callback передаёт adm_tags; stream-опции читаются из принятого userinfo после commit, +а не трактуются как поля adm_tags. +Merkle уже обеспечивает версии, подписи и обмен для поздних участников; второй механизм хранения объявлений не создаётся. +Rates, RTT и очереди запрашиваются прямо у кандидата, а не записываются в Merkle каждые 200 мс. +Входящий SUBSCRIBE проходит авторизацию по членству; задержка распространения его опции не должна создавать взаимное ожидание. +Сам получатель обязан иметь локальную подписку до запросов, а выдающий узел — свою до обработки потока. + +14. Жизненный цикл и диагностика + +При остановке последнего локального владельца прекращаем producers/поиск, снимаем опцию подписки, +отменяем callbacks/таймеры/waiters, закрываем сетевые выдачи, убираем задания и освобождаем свои NCD handles. +Кольцо и FEC-контекст освобождаются после прекращения обращений. +Работа других трансляций/модулей на общем ETCP сохраняется. +Удаление группы, исключение мембера и смена source_epoch отменяют относящиеся к ним выдачи. +Stop/destroy выполняются по правилам ядра, вне callbacks останавливаемого сервиса. + +Категория stream с конфигурируемым уровнем: +- INFO: запуск/остановка с FEC/размером/лимитами, выдачи, перенос, подтверждённая перегрузка и её результат, итоговые счётчики. +- DEBUG: выбор кандидата с метриками/причиной, окно кольца, timestamp-дельты/lag/rates, сводки очередей и циклов проб. +- TRACE: отдельные блоки, протокольные команды и важные ветви при включённой подробной диагностике. +- WARN/ERROR: отказные и нештатные ветки с контекстом, включая подпись, эпоху, LOOP, EXPIRED, память, аренду и транспорт. + +Указываются группа/источник/эпоха, peer, handle/subblock, индекс, состояние, причина и значимые величины. +При перегрузке — A/D/X/Q, возраст, длительность подтверждения, g, rates выбранных выдач и требуемое сокращение. +При timestamp-разрыве — предыдущий/новый индекс, timestamp и причина сброса базы. +Ключи и содержимое данных не выводятся; частые события агрегируются с полными счётчиками. +Не выдавать принятие в очередь за доставку, относительный lag за абсолютную задержку, старую опцию за онлайн-подписку. + +15. Реализация и проверки + +Порядок: +1) Подписки/экземпляры, member-опции и BGP-фильтр, дескриптор, FEC/подпись/timestamp, кольцо. +2) Raw-протокол, NCD + reverse, выдачи с индекса/конца, аренда и защита от повторов. +3) Общая очередь/backpressure, относительные метрики, rates и разгрузка. +4) Стартовые 32 пробы/последующий цикл, выбор путей, защита от циклов и перенос. +5) API непрозрачных данных и потребителя, без интеграции видеокодека в этом этапе. + +Проверки: +- Несколько подписок/источников/групп на одном ETCP; stop одной сохраняет остальные и чужое владение. +- Неподписанный узел не создаёт экземпляр, не выполняет crypto/FEC для DATA и не ретранслирует его. +- Источник задаёт различные FEC; decode из k проверенных блоков и невозможность при k-1. +- Подпись каждого исходного/проверочного блока; подмена payload/timestamp/индекса/FEC/эпохи/группы отвергается. +- Кодограммы около 1300 байт, полный/неполный буфер, источник сбрасывает за 100 мс, ретранслятор не имеет такого таймера. +- Оборот timestamp, нулевая дельта, постепенное отставание, догон, полная остановка и длинный разрыв без ложного числа оборотов. +- Переполнение 32-МБ кольца, дырки, EXPIRED до отправки, курсор медленного получателя и ограниченность памяти. +- Первый подписанный блок с конца задаёт общий N; остальные выдачи начинают с него, а не с независимых концов. +- Идемпотентные запросы/продления/закрытия, водяной знак handles, поздние сообщения, REINIT и новая эпоха источника. +- Reverse/неуспех/отмена; потеря одного линка сохраняет выдачи, DOWN всего NCD запускает восстановление. +- Перекрытие поставщиков без дублирования потребителю, rollback, одновременное восстановление нескольких субпотоков. +- Цикл по устаревшим метрикам, одновременная перестройка и выдача сохранённых путей: DATA не циркулирует. +- Глитч короче подтверждения, устойчивое отставание 500 мс, нулевой выход и локальная перегрузка uasync. +- Выбор числа закрытий по g/rates, разные rates, повтор OVERLOAD, частичное выполнение и закрытие остатка через 2 секунды. +- 32 совместимых кандидата на старте, далее 1/с, маленький/большой список и запрет повторного полного цикла раньше 30 с. +- Merkle-опции до/после BGP-события, офлайн-опция не кандидат, поздний участник, перезапуск и сохранение других полей userinfo. +- Мобильный источник/получатель и исключение мобильных из фоновых проб ретрансляторов. + +Перед кодированием фиксируются точная раскладка stream_proto.h, строковая схема member-опций и численные +лимиты собственной очереди/служебной памяти. Это детализация реализации принятых решений, а не повторный выбор транспорта, +подписи, часов, формата содержимого, анонсов, отдельного дозапроса или расписания проб. +Гистерезис обычного переноса и небольшой запас rate сверх g подбираются на стенде с сохранением заданных сроков. + +Исходники добавляются в Autotools и применимые desktop/Android CMake-сборки общего ядра. +Тесты добавляются через tests/Makefile.am. Диагностика должна однозначно показывать выбор путей, +пропуски, сбросы timestamp-базы, переключения и фактический результат разгрузки. + +16. План работ и текущий статус реализации + +Реализация начата 10.10.2026. Работа разбита на проверяемые этапы: + +[+] Основа данных в src/stream: + stream_proto.h — канонический big-endian формат подписанного блока; дефолтный полный блок 1300 байт. + stream_block — RS и Ed25519-подпись каждого блока; проверка ключа трансляции, параметров и содержимого. + stream_buffer — общий бюджет 32 МБ по фактически сохранённым блокам, индекс пачек и вытеснение старейшей пачки. + stream_metrics — относительное отставание, rate, timestamp wrap/reset и минутная история проблем/остановок. + stream_source — приём непрозрачных байтов, немедленная выдача полной пачки, таймер остатка 100 мс и отмена при destroy. + Добавлены категория stream, Autotools-сборка и test_stream_core. + Desktop/Android CMake уже используют GLOB_RECURSE для src/*.c и автоматически включают новые компоненты. + Проверки: подписи и подмены полей, все 28 пар потерь при 6+2, другие FEC-параметры, padding, + дырки/переполнение кольца, timestamp через ноль/длинный перерыв, остановка и отмена таймеров, учёт утечек. + ./build.sh --no-gui -j4 и ./check.sh прошли: 143 passed, 0 failed, 0 skipped; + отдельные proxy/full, burst и load также PASS. Лог: /tmp/stream-check.log. + +[ ] Включение в жизненный цикл чат-сервиса: + контекст модуля в UTUN_INSTANCE, публичные локальные подписки/публикация, события потребителя, + stream_source/stream_subscriptions в блоке мембера, сохранение остальных userinfo и BGP-фильтр кандидатов. + +[ ] Сетевая выдача: + raw/router IDs, INFO и поколения сессий, 64-битные handles, SUBSCRIBE/RENEW/CLOSED, + собственное владение NCD, REVERSE_REQ, курсоры кольца и общий начальный индекс от первого поставщика. + +[ ] Планировщик отправки и разгрузка: + одна очередь на пира, backpressure, ограничение памяти/возраста, A/D/X/Q, + подтверждение 500 мс, сокращение по g/rates и двухсекундный срок исполнения OVERLOAD. + +[ ] Оптимизация сети и интеграционные проверки: + пути/защита от циклов, стартовые 32 пробы, дальнейший темп 1/с и полный цикл не чаще 30 с, + постепенный перенос с перекрытием, failover, несколько источников и независимое закрытие на общем ETCP. + +Текущие компоненты — основа модуля данных; сетевые bindings, Merkle-опции и вызовы из UTUN_INSTANCE ещё не подключены. diff --git a/doc/tasks.md b/doc/tasks.md index a924b77f..ba9b2f53 100644 --- a/doc/tasks.md +++ b/doc/tasks.md @@ -4,6 +4,14 @@ ## Текущие задачи +[ ] **Стриминг: реализация по stream.txt** — 10.10.2026: + ТЗ и этапы: doc/stream.txt, разделы 15–16. Основа данных реализована в src/stream: + подписанные FEC-блоки 1300 байт, общий буфер 32 МБ, относительные timestamp-метрики, + источник с таймером 100 мс и категория stream. test_stream_core прошёл; + ./check.sh: 143 passed, 0 failed, 0 skipped; proxy/full, burst и load — PASS. + Следующий этап — жизненный цикл CHAT и опции мембера через Merkle с фильтром BGP, + затем NCD-выдачи/reverse, общий отправитель/перегрузка и оптимизация путей. + [+] **test_reality_tls: коллизии порта слушателя** — 10.10.2026: В общем прогоне 140 passed, 1 failed: stcp_server_create не смог занять порт 34922, bind errno=98 (EADDRINUSE). Причина: выбор порта как 34000 + time(NULL) % 1000 без гарантии его доступности. diff --git a/lib/debug_config.c b/lib/debug_config.c index e4e6908d..1762246d 100644 --- a/lib/debug_config.c +++ b/lib/debug_config.c @@ -136,6 +136,7 @@ static const struct { {"radio", DEBUG_CATEGORY_RADIO}, {"vad", DEBUG_CATEGORY_VAD}, {"aec", DEBUG_CATEGORY_AEC}, + {"stream", DEBUG_CATEGORY_STREAM}, {"all", DEBUG_CATEGORY_ALL}, {NULL, DEBUG_CATEGORY_NONE} }; diff --git a/lib/debug_config.h b/lib/debug_config.h index 0f48edee..fa61be6a 100644 --- a/lib/debug_config.h +++ b/lib/debug_config.h @@ -74,7 +74,8 @@ typedef int debug_category_t; #define DEBUG_CATEGORY_RADIO 33 // Radio — PTT walkie-talkie multicast (voice) #define DEBUG_CATEGORY_VAD 34 // VAD — Silero voice activity detector (ONNX Runtime) #define DEBUG_CATEGORY_AEC 35 // AEC — acoustic echo cancellation (SpeexDSP) -#define DEBUG_CATEGORY_COUNT 36 // Total number of categories +#define DEBUG_CATEGORY_STREAM 36 // Подписанные потоки, FEC, буферы и выбор поставщиков +#define DEBUG_CATEGORY_COUNT 37 // Total number of categories #define DEBUG_CATEGORY_ALL (-1) // special value for all categories /* Debug configuration structure */ diff --git a/src/Makefile.am b/src/Makefile.am index a8ec43e9..37779348 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -3,8 +3,20 @@ noinst_LIBRARIES = libutun.a EXTRA_DIST = lwip_tcp/COPYING dnsmasq/COPYING dnsmasq/COPYING-v3 +stream_core_files = \ + stream/stream_proto.h \ + stream/stream_block.h \ + stream/stream_block.c \ + stream/stream_buffer.h \ + stream/stream_buffer.c \ + stream/stream_metrics.h \ + stream/stream_metrics.c \ + stream/stream_source.h \ + stream/stream_source.c + # Sources that are always compiled utun_CORE_SOURCES = \ + $(stream_core_files) \ utun.c \ utun_instance.c \ config_parser.c \ @@ -152,6 +164,7 @@ utun_CORE_SOURCES = \ # libutun: all core sources except main() libutun_a_SOURCES = \ + $(stream_core_files) \ utun_instance.c \ config_parser.c \ config_updater.c \ diff --git a/src/stream/stream_block.c b/src/stream/stream_block.c new file mode 100644 index 00000000..2b3f9ce8 --- /dev/null +++ b/src/stream/stream_block.c @@ -0,0 +1,180 @@ +#include + +#include +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "stream_block.h" + +struct stream_encoder { + struct stream_descriptor descriptor; + struct rs_codec *rs; + uint8_t private_key[SC_PRIVKEY_SIZE]; + uint8_t public_key[SC_PUBKEY_SIZE]; + uint8_t *input; + uint8_t *wire; +}; + +/* Проверить ключ трансляции и параметры до расчёта размеров/выделения памяти. */ +int stream_descriptor_validate(const struct stream_descriptor *d) { + if (!d || !d->group_id || !d->source_node_id || !d->source_epoch || !d->n || d->r >= d->n + || d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "descriptor invalid: present=%d group=%llx source=%llx epoch=%llx n=%u r=%u size=%u", + d != NULL, d ? (unsigned long long)d->group_id : 0, d ? (unsigned long long)d->source_node_id : 0, + d ? (unsigned long long)d->source_epoch : 0, d ? d->n : 0, d ? d->r : 0, d ? d->block_size : 0); + return STREAM_ERR_ARGUMENT; + } + return STREAM_OK; +} + +size_t stream_block_size(const struct stream_descriptor *d) { + return stream_descriptor_validate(d) == STREAM_OK ? d->block_size + STREAM_BLOCK_OVERHEAD : 0; +} + +size_t stream_input_size(const struct stream_descriptor *d) { + return stream_descriptor_validate(d) == STREAM_OK ? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) : 0; +} + +/* Создать рабочую память RS и получить публичный ключ именно переданного Ed25519 seed. */ +int stream_encoder_create(struct stream_encoder **out, const struct stream_descriptor *d, const uint8_t private_key[32]) { + if (!out || *out || !private_key || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder create invalid arguments out=%p key=%p", (void *)out, (const void *)private_key); + return STREAM_ERR_ARGUMENT; + } + struct stream_encoder *encoder = u_calloc(1, sizeof(*encoder)); + if (!encoder) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder allocation failed source=%llx", (unsigned long long)d->source_node_id); + return STREAM_ERR_MEMORY; + } + encoder->descriptor = *d; + memcpy(encoder->private_key, private_key, sizeof(encoder->private_key)); + EVP_PKEY *key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, private_key, SC_PRIVKEY_SIZE); + size_t public_length = SC_PUBKEY_SIZE; + int key_ok = key && EVP_PKEY_get_raw_public_key(key, encoder->public_key, &public_length) == 1 && public_length == SC_PUBKEY_SIZE; + EVP_PKEY_free(key); + if (!key_ok) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder Ed25519 key initialization failed source=%llx", + (unsigned long long)d->source_node_id); + stream_encoder_destroy(encoder); + return STREAM_ERR_CRYPTO; + } + if (rs_init(&encoder->rs, d->n, d->r, d->block_size) != RS_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder RS initialization failed n=%u r=%u size=%u", d->n, d->r, d->block_size); + stream_encoder_destroy(encoder); + return STREAM_ERR_FEC; + } + encoder->input = u_malloc((uint32_t)stream_input_size(d)); + encoder->wire = u_malloc((uint32_t)(d->n * stream_block_size(d))); + if (!encoder->input || !encoder->wire) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder working memory allocation failed n=%u size=%u", d->n, d->block_size); + stream_encoder_destroy(encoder); + return STREAM_ERR_MEMORY; + } + *out = encoder; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder created group=%llx source=%llx epoch=%llx FEC=%u+%u block=%zu input=%zu", + (unsigned long long)d->group_id, (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, + d->n - d->r, d->r, stream_block_size(d), stream_input_size(d)); + return STREAM_OK; +} + +/* Очистить секретный seed и рабочее содержимое до возврата памяти allocator. */ +void stream_encoder_destroy(struct stream_encoder *encoder) { + if (!encoder) return; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder destroy source=%llx epoch=%llx", + (unsigned long long)encoder->descriptor.source_node_id, (unsigned long long)encoder->descriptor.source_epoch); + OPENSSL_cleanse(encoder->private_key, sizeof(encoder->private_key)); + if (encoder->input) OPENSSL_cleanse(encoder->input, stream_input_size(&encoder->descriptor)); + u_free(encoder->input); + u_free(encoder->wire); + rs_free(encoder->rs); + u_free(encoder); +} + +const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder) { + if (!encoder) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder public key: null context"); return NULL; } + return encoder->public_key; +} + +/* Сформировать канонический заголовок, RS-кодограммы и подпись каждого блока. */ +int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags, + const uint8_t *input, size_t length, const uint8_t **blocks) { + if (!encoder || !input || !blocks || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET) + || length > stream_input_size(&encoder->descriptor)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode invalid arguments seq=%llu len=%zu flags=%u", (unsigned long long)seq, length, flags); + return STREAM_ERR_ARGUMENT; + } + const struct stream_descriptor *d = &encoder->descriptor; + size_t wire_size = stream_block_size(d); + uint8_t *rs_blocks[RS_MAX_BLOCKS]; + memcpy(encoder->input, input, length); + memset(encoder->input + length, 0, stream_input_size(d) - length); + for (unsigned i = 0; i < d->n; i++) { + uint8_t *wire = encoder->wire + i * wire_size; + memcpy(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE); + stream_write_be(wire + STREAM_BLOCK_GROUP_OFF, d->group_id, 8); + stream_write_be(wire + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8); + stream_write_be(wire + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8); + wire[STREAM_BLOCK_N_OFF] = d->n; + wire[STREAM_BLOCK_R_OFF] = d->r; + stream_write_be(wire + STREAM_BLOCK_SIZE_OFF, d->block_size, 2); + stream_write_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2); + stream_write_be(wire + STREAM_BLOCK_LENGTH_OFF, length, 4); + wire[STREAM_BLOCK_FLAGS_OFF] = flags; + rs_blocks[i] = wire + STREAM_BLOCK_RS_OFF; + } + if (rs_encode(encoder->rs, seq, encoder->input, rs_blocks) != RS_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode RS failed seq=%llu", (unsigned long long)seq); + return STREAM_ERR_FEC; + } + for (unsigned i = 0; i < d->n; i++) { + uint8_t *wire = encoder->wire + i * wire_size; + if (sc_ed25519_sign(encoder->private_key, wire, wire_size - SC_SIGN_SIZE, wire + wire_size - SC_SIGN_SIZE) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode signature failed seq=%llu sub=%u", (unsigned long long)seq, i); + return STREAM_ERR_CRYPTO; + } + } + for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * wire_size; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded seq=%llu timestamp=%u len=%zu n=%u flags=%u", + (unsigned long long)seq, timestamp_ms, length, d->n, flags); + return STREAM_OK; +} + +/* Проверить размеры, область подписи, дескриптор, метаданные и подпись до публикации view. */ +int stream_block_verify(const struct stream_descriptor *d, const uint8_t public_key[32], const uint8_t *wire, + size_t length, struct stream_block_view *view) { + if (!public_key || !wire || !view || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "verify invalid arguments length=%zu", length); + return STREAM_ERR_ARGUMENT; + } + if (length != stream_block_size(d) || memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE) + || stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id + || stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id + || stream_read_be(wire + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch + || wire[STREAM_BLOCK_N_OFF] != d->n || wire[STREAM_BLOCK_R_OFF] != d->r + || stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != d->block_size) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify format/descriptor mismatch source=%llx epoch=%llx length=%zu expected=%zu", + (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_size(d)); + return STREAM_ERR_FORMAT; + } + struct stream_block_view parsed = { + .seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8), + .data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4), + .timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), + .subblock = wire[STREAM_BLOCK_SUB_OFF], .flags = wire[STREAM_BLOCK_FLAGS_OFF], .rs = wire + STREAM_BLOCK_RS_OFF + }; + if (parsed.seq == UINT64_MAX || parsed.subblock >= d->n || !parsed.data_length + || parsed.data_length > stream_input_size(d) || (parsed.flags & ~STREAM_BLOCK_TIME_RESET)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify invalid metadata seq=%llu sub=%u length=%u flags=%u", + (unsigned long long)parsed.seq, parsed.subblock, parsed.data_length, parsed.flags); + return STREAM_ERR_FORMAT; + } + if (sc_ed25519_verify(public_key, wire, length - SC_SIGN_SIZE, wire + length - SC_SIGN_SIZE) != SC_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify source signature failed source=%llx epoch=%llx seq=%llu sub=%u", + (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, + (unsigned long long)parsed.seq, parsed.subblock); + return STREAM_ERR_SIGNATURE; + } + *view = parsed; + return STREAM_OK; +} diff --git a/src/stream/stream_block.h b/src/stream/stream_block.h new file mode 100644 index 00000000..4822b60e --- /dev/null +++ b/src/stream/stream_block.h @@ -0,0 +1,48 @@ +/* Формирование и проверка подписанных FEC-блоков без сокетов и внешних callbacks. + * Контексты одного uasync-потока; descriptor и ключ источника уже проверены членством группы. + * encode возвращает заимствованные блоки до следующего encode/destroy; ошибочные результаты не публикуют. + * verify не меняет выходной view при отказе. Ключ непосредственного ретранслятора здесь не используется. */ +#ifndef STREAM_BLOCK_H +#define STREAM_BLOCK_H + +#include "stream_proto.h" + +#ifdef __cplusplus +extern "C" { +#endif + +struct stream_encoder; + +struct stream_block_view { + uint64_t seq; + uint32_t data_length; + uint16_t timestamp_ms; + uint8_t subblock; + uint8_t flags; + const uint8_t *rs; /* Заимствованный диапазон descriptor.block_size байт. */ +}; + +int stream_descriptor_validate(const struct stream_descriptor *descriptor); +size_t stream_block_size(const struct stream_descriptor *descriptor); +size_t stream_input_size(const struct stream_descriptor *descriptor); + +/* out заранее NULL; ключ копируется и очищается при destroy. */ +int stream_encoder_create(struct stream_encoder **out, const struct stream_descriptor *descriptor, + const uint8_t private_key[SC_PRIVKEY_SIZE]); +void stream_encoder_destroy(struct stream_encoder *encoder); +/* Заимствованный публичный ключ, полученный из private_key этого encoder. */ +const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder); + +/* blocks — массив descriptor.n указателей; input не перекрывается с рабочими буферами encoder. + * 0 < length <= stream_input_size; остаток пачки дополняется нулями. seq != UINT64_MAX. + * При ошибке blocks не изменяется; собственные рабочие буферы могут измениться. */ +int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags, + const uint8_t *input, size_t length, const uint8_t **blocks); + +int stream_block_verify(const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE], + const uint8_t *wire, size_t length, struct stream_block_view *view); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c new file mode 100644 index 00000000..95698aaa --- /dev/null +++ b/src/stream/stream_buffer.c @@ -0,0 +1,330 @@ +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "stream_buffer.h" + +#define STREAM_SLOT_NONE UINT32_MAX + +struct stream_cached_slot { + uint32_t next; + uint8_t subblock; +}; + +struct stream_cached_batch { + struct stream_cached_batch *hash_next; + uint64_t seq; + uint32_t first_slot; + uint32_t data_length; + uint16_t timestamp_ms; + uint16_t count; + uint8_t flags; +}; + +struct stream_buffer { + struct stream_descriptor descriptor; + uint8_t public_key[SC_PUBKEY_SIZE]; + struct rs_codec *rs; + uint8_t *decode_output; + uint8_t *data; + struct stream_cached_slot *slots; + struct stream_cached_batch **hash; + struct stream_cached_batch **heap; + uint64_t latest[RS_MAX_BLOCKS]; + uint8_t latest_valid[RS_MAX_BLOCKS]; + size_t block_bytes; + uint32_t capacity; + uint32_t hash_size; + uint32_t free_slot; + struct stream_buffer_stats stats; +}; + +/* Индекс пачки не зависит от порядка поступления; множество дырок не требует линейного поиска. */ +static uint32_t buffer_hash(const struct stream_buffer *b, uint64_t seq) { + seq ^= seq >> 33; + seq *= UINT64_C(0xff51afd7ed558ccd); + seq ^= seq >> 33; + return (uint32_t)seq & (b->hash_size - 1); +} + +static struct stream_cached_batch *buffer_batch(const struct stream_buffer *b, uint64_t seq) { + for (struct stream_cached_batch *batch = b->hash[buffer_hash(b, seq)]; batch; batch = batch->hash_next) + if (batch->seq == seq) return batch; + return NULL; +} + +/* Поиск не более n блоков пачки; пустые субпотоки не занимают отдельные позиции в массиве данных. */ +static uint32_t buffer_slot(const struct stream_buffer *b, const struct stream_cached_batch *batch, uint8_t subblock) { + for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next) + if (b->slots[slot].subblock == subblock) return slot; + return STREAM_SLOT_NONE; +} + +/* Обмен узлов min-heap, упорядоченного по индексу пачки. */ +static void buffer_heap_swap(struct stream_buffer *b, uint32_t first, uint32_t second) { + struct stream_cached_batch *batch = b->heap[first]; + b->heap[first] = b->heap[second]; + b->heap[second] = batch; +} + +/* Добавить новую пачку в хеш и heap; память уже выделена, отказов здесь нет. */ +static void buffer_insert_batch(struct stream_buffer *b, struct stream_cached_batch *batch) { + uint32_t hash = buffer_hash(b, batch->seq); + batch->hash_next = b->hash[hash]; + b->hash[hash] = batch; + uint32_t index = b->stats.batches++; + b->heap[index] = batch; + while (index && b->heap[(index - 1) / 2]->seq > batch->seq) { + buffer_heap_swap(b, index, (index - 1) / 2); + index = (index - 1) / 2; + } +} + +/* Удалить целую минимальную пачку; выданные наружу указатели становятся недействительными. */ +static void buffer_evict_oldest(struct stream_buffer *b) { + struct stream_cached_batch *batch = b->heap[0]; + uint32_t hash = buffer_hash(b, batch->seq); + struct stream_cached_batch **link = &b->hash[hash]; + while (*link != batch) link = &(*link)->hash_next; + *link = batch->hash_next; + uint32_t remaining = --b->stats.batches; + if (remaining) { + b->heap[0] = b->heap[remaining]; + uint32_t index = 0; + while (index * 2 + 1 < remaining) { + uint32_t child = index * 2 + 1; + if (child + 1 < remaining && b->heap[child + 1]->seq < b->heap[child]->seq) child++; + if (b->heap[index]->seq <= b->heap[child]->seq) break; + buffer_heap_swap(b, index, child); + index = child; + } + } + for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE;) { + uint32_t next = b->slots[slot].next; + unsigned sub = b->slots[slot].subblock; + if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0; + b->slots[slot].next = b->free_slot; + b->free_slot = slot; + slot = next; + } + b->stats.expired_before = batch->seq + 1; + b->stats.blocks -= batch->count; + b->stats.stored_bytes -= batch->count * b->block_bytes; + b->stats.evicted_batches++; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "buffer evicted seq=%llu blocks=%u retained=%u bytes=%zu low=%llu", + (unsigned long long)batch->seq, batch->count, b->stats.blocks, b->stats.stored_bytes, + (unsigned long long)b->stats.expired_before); + u_free(batch); +} + +/* Выделить ограниченные массивы данных/индексов; память пачек растёт только по фактическому поступлению. */ +int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *d, const uint8_t public_key[32], size_t bytes) { + if (!out || *out || !public_key || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer create invalid arguments out=%p key=%p", (void *)out, (const void *)public_key); + return STREAM_ERR_ARGUMENT; + } + size_t block_bytes = stream_block_size(d); + if (!bytes) bytes = STREAM_BUFFER_BYTES; + if (bytes > STREAM_BUFFER_BYTES || bytes < d->n * block_bytes) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer budget invalid bytes=%zu batch=%zu max=%u", bytes, d->n * block_bytes, STREAM_BUFFER_BYTES); + return STREAM_ERR_ARGUMENT; + } + struct stream_buffer *b = u_calloc(1, sizeof(*b)); + if (!b) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer context allocation failed bytes=%zu", bytes); return STREAM_ERR_MEMORY; } + b->descriptor = *d; + memcpy(b->public_key, public_key, SC_PUBKEY_SIZE); + b->block_bytes = block_bytes; + b->capacity = (uint32_t)(bytes / block_bytes); + b->hash_size = 1; + while (b->hash_size < b->capacity) b->hash_size *= 2; + b->stats.capacity_bytes = b->capacity * block_bytes; + b->data = u_malloc((uint32_t)b->stats.capacity_bytes); + b->slots = u_calloc(b->capacity, sizeof(*b->slots)); + b->heap = u_calloc(b->capacity, sizeof(*b->heap)); + b->hash = u_calloc(b->hash_size, sizeof(*b->hash)); + b->decode_output = u_malloc((uint32_t)stream_input_size(d)); + if (!b->data || !b->slots || !b->heap || !b->hash || !b->decode_output) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer arrays allocation failed capacity=%u bytes=%zu", b->capacity, b->stats.capacity_bytes); + stream_buffer_destroy(b); + return STREAM_ERR_MEMORY; + } + for (uint32_t i = 0; i < b->capacity; i++) b->slots[i].next = i + 1; + b->slots[b->capacity - 1].next = STREAM_SLOT_NONE; + if (rs_init(&b->rs, d->n, d->r, d->block_size) != RS_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS initialization failed n=%u r=%u", d->n, d->r); + stream_buffer_destroy(b); + return STREAM_ERR_FEC; + } + *out = b; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer created source=%llx epoch=%llx bytes=%zu block=%zu slots=%u", + (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, + b->stats.capacity_bytes, b->block_bytes, b->capacity); + return STREAM_OK; +} + +/* Прекращение владельцем всех обращений предшествует destroy; кольцо ничего не знает о чужих NCD handles. */ +void stream_buffer_destroy(struct stream_buffer *b) { + if (!b) return; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu", + (unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes, + (unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates); + for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]); + rs_free(b->rs); + u_free(b->decode_output); + u_free(b->data); + u_free(b->slots); + u_free(b->heap); + u_free(b->hash); + u_free(b); +} + +/* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */ +int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) { + if (!b || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer put invalid arguments length=%zu", length); return STREAM_ERR_ARGUMENT; } + struct stream_cached_batch *batch = NULL; + if (length == b->block_bytes) { + batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8)); + uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; + if (batch && sub < b->descriptor.n) { + uint32_t slot = buffer_slot(b, batch, sub); + if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) { + b->stats.duplicates++; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, sub); + return STREAM_DUPLICATE; + } + } + } + struct stream_block_view view; + int result = stream_block_verify(&b->descriptor, b->public_key, wire, length, &view); + if (result != STREAM_OK) return result; + if (view.seq < b->stats.expired_before) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer expired incoming seq=%llu sub=%u low=%llu full=%d", + (unsigned long long)view.seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity); + return STREAM_EXPIRED; + } + if (batch && (batch->timestamp_ms != view.timestamp_ms || batch->data_length != view.data_length || batch->flags != view.flags + || buffer_slot(b, batch, view.subblock) != STREAM_SLOT_NONE)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer source conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u", + (unsigned long long)view.seq, view.subblock, batch->timestamp_ms, view.timestamp_ms, + batch->data_length, view.data_length, batch->flags, view.flags); + return STREAM_ERR_CONFLICT; + } + if (b->stats.blocks == b->capacity && view.seq <= b->heap[0]->seq) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer full: incoming seq=%llu sub=%u cannot replace newer data oldest=%llu", + (unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq); + return STREAM_EXPIRED; + } + int new_batch = batch == NULL; + if (new_batch) { + batch = u_calloc(1, sizeof(*batch)); + if (!batch) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq); + return STREAM_ERR_MEMORY; + } + batch->seq = view.seq; + batch->first_slot = STREAM_SLOT_NONE; + batch->timestamp_ms = view.timestamp_ms; + batch->data_length = view.data_length; + batch->flags = view.flags; + } + if (b->stats.blocks == b->capacity) buffer_evict_oldest(b); + if (new_batch) buffer_insert_batch(b, batch); + uint32_t slot = b->free_slot; + b->free_slot = b->slots[slot].next; + b->slots[slot].subblock = view.subblock; + b->slots[slot].next = batch->first_slot; + batch->first_slot = slot; + batch->count++; + memcpy(b->data + slot * b->block_bytes, wire, length); + b->stats.blocks++; + b->stats.stored_bytes += length; + if (view.seq > b->stats.highest_seq) b->stats.highest_seq = view.seq; + if (!b->latest_valid[view.subblock] || view.seq > b->latest[view.subblock]) { + b->latest[view.subblock] = view.seq; + b->latest_valid[view.subblock] = 1; + } + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer stored seq=%llu sub=%u present=%u/%u bytes=%zu", + (unsigned long long)view.seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes); + return STREAM_OK; +} + +/* Найти точную позицию либо последний блок субпотока, не меняя его историю/срок хранения. */ +int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock, + const uint8_t **wire, size_t *length, uint64_t *actual_seq) { + if (!b || !wire || !length || !actual_seq || subblock >= b->descriptor.n) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer read invalid arguments seq=%llu sub=%u", (unsigned long long)seq, subblock); + return STREAM_ERR_ARGUMENT; + } + if (seq == UINT64_MAX) { + if (!b->latest_valid[subblock]) return STREAM_WAIT; + seq = b->latest[subblock]; + } + if (seq < b->stats.expired_before) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "buffer read expired seq=%llu low=%llu", (unsigned long long)seq, + (unsigned long long)b->stats.expired_before); + return STREAM_EXPIRED; + } + if (!b->stats.blocks || seq > b->stats.highest_seq) return STREAM_WAIT; + struct stream_cached_batch *batch = buffer_batch(b, seq); + uint32_t slot = batch ? buffer_slot(b, batch, subblock) : STREAM_SLOT_NONE; + if (slot == STREAM_SLOT_NONE) { + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer read gap seq=%llu sub=%u", (unsigned long long)seq, subblock); + return STREAM_GAP; + } + *wire = b->data + slot * b->block_bytes; + *length = b->block_bytes; + *actual_seq = seq; + return STREAM_OK; +} + +/* Границы наличия конкретного субпотока не обещают заполнения всех промежуточных индексов. */ +int stream_buffer_range(struct stream_buffer *b, uint8_t subblock, uint64_t *first, uint64_t *last) { + if (!b || !first || !last || subblock >= b->descriptor.n) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer range invalid sub=%u", subblock); + return STREAM_ERR_ARGUMENT; + } + if (!b->latest_valid[subblock]) return STREAM_WAIT; + uint64_t low = b->latest[subblock]; + for (uint32_t i = 0; i < b->stats.batches; i++) { + struct stream_cached_batch *batch = b->heap[i]; + if (batch->seq < low && buffer_slot(b, batch, subblock) != STREAM_SLOT_NONE) low = batch->seq; + if (low == b->heap[0]->seq) break; + } + *first = low; + *last = b->latest[subblock]; + return STREAM_OK; +} + +/* RS пишет в собственный scratch; ошибка или недостающие блоки не повреждают выход потребителя. */ +int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output, size_t capacity, size_t *length) { + if (!b || !output || !length || seq == UINT64_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode invalid arguments seq=%llu capacity=%zu", (unsigned long long)seq, capacity); + return STREAM_ERR_ARGUMENT; + } + if (seq < b->stats.expired_before) return STREAM_EXPIRED; + struct stream_cached_batch *batch = buffer_batch(b, seq); + if (!batch) return !b->stats.blocks || seq > b->stats.highest_seq ? STREAM_WAIT : STREAM_GAP; + if (batch->count < b->descriptor.n - b->descriptor.r) return STREAM_WAIT; + if (capacity < batch->data_length) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode output too small seq=%llu capacity=%zu needed=%u", + (unsigned long long)seq, capacity, batch->data_length); + return STREAM_ERR_ARGUMENT; + } + const uint8_t *blocks[RS_MAX_BLOCKS] = {0}; + for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next) + blocks[b->slots[slot].subblock] = b->data + slot * b->block_bytes + STREAM_BLOCK_RS_OFF; + if (rs_decode(b->rs, blocks, b->decode_output) != RS_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS failed seq=%llu present=%u", (unsigned long long)seq, batch->count); + return STREAM_ERR_FEC; + } + memcpy(output, b->decode_output, batch->data_length); + *length = batch->data_length; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer decoded seq=%llu present=%u length=%u", (unsigned long long)seq, batch->count, batch->data_length); + return STREAM_OK; +} + +int stream_buffer_get_stats(const struct stream_buffer *b, struct stream_buffer_stats *stats) { + if (!b || !stats) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer stats invalid arguments"); return STREAM_ERR_ARGUMENT; } + *stats = b->stats; + return STREAM_OK; +} diff --git a/src/stream/stream_buffer.h b/src/stream/stream_buffer.h new file mode 100644 index 00000000..47445208 --- /dev/null +++ b/src/stream/stream_buffer.h @@ -0,0 +1,53 @@ +/* Кольцо проверенных блоков одного запуска: общий бюджет данных для всех субпотоков. + * Пустые субпотоки не резервируют блоки: 32 МБ занимают реально сохранённые кодограммы. + * Вытеснение целой старейшей пачки не закрепляет данные для медленного получателя. + * Все вызовы — один uasync-поток. read возвращает память до следующего put/destroy. + * public_key должен происходить из проверенной записи источника в группе. */ +#ifndef STREAM_BUFFER_H +#define STREAM_BUFFER_H + +#include "stream_block.h" + +#ifdef __cplusplus +extern "C" { +#endif + +struct stream_buffer; + +struct stream_buffer_stats { + size_t capacity_bytes; + size_t stored_bytes; + uint32_t blocks; + uint32_t batches; + uint64_t expired_before; + uint64_t highest_seq; + uint64_t evicted_batches; + uint64_t duplicates; +}; + +/* out заранее NULL; bytes=0 выбирает 32 МБ, меньший бюджет допускается для стенда. + * Должна помещаться хотя бы одна полная пачка n блоков; верхний предел — STREAM_BUFFER_BYTES. */ +int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *descriptor, + const uint8_t public_key[SC_PUBKEY_SIZE], size_t bytes); +void stream_buffer_destroy(struct stream_buffer *buffer); + +/* Новый блок проверяется до изменения кольца. Повтор неизменного блока: STREAM_DUPLICATE. + * STREAM_EXPIRED не меняет более свежие данные; неверная подпись/конфликт возвращает отрицательный код. */ +int stream_buffer_put(struct stream_buffer *buffer, const uint8_t *wire, size_t length); + +/* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока. + * WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен. + * При отсутствии блока выходные указатели/номер не меняются. */ +int stream_buffer_read(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, + const uint8_t **wire, size_t *length, uint64_t *actual_seq); +int stream_buffer_range(struct stream_buffer *buffer, uint8_t subblock, uint64_t *first, uint64_t *last); + +/* Восстановить пачку из проверенных блоков; достаточно capacity >= действительной длины. + * Выход не меняется при WAIT/GAP/EXPIRED/ошибке; восстановленные отсутствующие блоки в кольцо не добавляются. */ +int stream_buffer_decode(struct stream_buffer *buffer, uint64_t seq, uint8_t *output, size_t capacity, size_t *length); +int stream_buffer_get_stats(const struct stream_buffer *buffer, struct stream_buffer_stats *stats); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/src/stream/stream_metrics.c b/src/stream/stream_metrics.c new file mode 100644 index 00000000..fb59715b --- /dev/null +++ b/src/stream/stream_metrics.c @@ -0,0 +1,136 @@ +#include + +#include "../../lib/debug_config.h" +#include "stream_metrics.h" + +/* Единственная корзина секундного интервала; old данные не переживают оборот окна. */ +static struct stream_metric_bucket *metrics_bucket(struct stream_metrics *m, uint64_t now_ms) { + uint64_t second = now_ms / 1000; + struct stream_metric_bucket *bucket = &m->buckets[second % STREAM_METRIC_SECONDS]; + if (!bucket->valid || bucket->second != second) { + memset(bucket, 0, sizeof(*bucket)); + bucket->second = second; + bucket->valid = 1; + } + return bucket; +} + +/* Переполнение вычисленного отставания не может превращать длительный stall в нулевую задержку. */ +static uint64_t metrics_add(uint64_t first, uint64_t second) { + return second > UINT64_MAX - first ? UINT64_MAX : first + second; +} + +/* Фиксировать одну остановку только после отсутствия нового блока сверх ожидаемого интервала + 500 мс. */ +static void metrics_detect_stop(struct stream_metrics *m, uint64_t now_ms) { + if (!m->initialized || m->ended || m->stopped) return; + uint64_t threshold = metrics_add(m->expected_interval_ms, STREAM_LATE_MS); + uint64_t elapsed = now_ms - m->last_receive_ms; + if (elapsed <= threshold) return; + uint64_t started = metrics_add(m->last_receive_ms, metrics_add(threshold, 1)); + metrics_bucket(m, started)->stops++; + m->stopped = 1; + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "subflow stopped after seq=%llu elapsed=%llums expected=%llums lag=%llums", + (unsigned long long)m->last_seq, (unsigned long long)elapsed, + (unsigned long long)m->expected_interval_ms, (unsigned long long)m->lag_ms); +} + +/* Начальная ожидаемая пауза — source flush 100 мс; после замеров используется фактический темп источника. */ +void stream_metrics_init(struct stream_metrics *m) { + if (!m) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics init null context"); return; } + memset(m, 0, sizeof(*m)); + m->expected_interval_ms = STREAM_SOURCE_FLUSH_TB / 10; +} + +/* Дельты относительно предшествующего блока исключают смещение часов и учитывают отставание по всей цепочке. */ +int stream_metrics_observe(struct stream_metrics *m, uint64_t seq, uint16_t timestamp_ms, uint8_t flags, + uint32_t bytes, uint64_t now_ms) { + if (!m || seq == UINT64_MAX || !bytes || (flags & ~STREAM_BLOCK_TIME_RESET)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics observe invalid seq=%llu bytes=%u flags=%u", (unsigned long long)seq, bytes, flags); + return STREAM_ERR_ARGUMENT; + } + if (m->ended || (m->initialized && now_ms < m->last_receive_ms)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "metrics state/clock invalid seq=%llu now=%llu previous=%llu ended=%d", + (unsigned long long)seq, (unsigned long long)now_ms, (unsigned long long)m->last_receive_ms, m->ended); + return STREAM_ERR_STATE; + } + if (m->initialized && seq <= m->last_seq) return STREAM_DUPLICATE; + metrics_detect_stop(m, now_ms); + uint64_t receive_delta = m->initialized ? now_ms - m->last_receive_ms : 0; + int rebase = !m->initialized || (flags & STREAM_BLOCK_TIME_RESET) || seq != m->last_seq + 1 + || receive_delta >= STREAM_TIMESTAMP_PERIOD_MS; + uint16_t source_delta = (uint16_t)(timestamp_ms - m->last_timestamp_ms); + struct stream_metric_bucket *bucket = metrics_bucket(m, now_ms); + bucket->blocks++; + if (rebase) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "metrics rebase seq=%llu previous=%llu timestamp=%u/%u receive_delta=%llu flags=%u", + (unsigned long long)seq, (unsigned long long)m->last_seq, m->last_timestamp_ms, timestamp_ms, + (unsigned long long)receive_delta, flags); + m->lag_ms = 0; + m->expected_interval_ms = STREAM_SOURCE_FLUSH_TB / 10; + } else { + if (receive_delta >= source_delta) m->lag_ms = metrics_add(m->lag_ms, receive_delta - source_delta); + else m->lag_ms = source_delta - receive_delta >= m->lag_ms ? 0 : m->lag_ms - (source_delta - receive_delta); + bucket->bytes += bytes; + bucket->source_ms += source_delta; + bucket->receive_ms += receive_delta; + bucket->intervals++; + if (bucket->source_ms) { + m->expected_interval_ms = bucket->source_ms / bucket->intervals; + if (!m->expected_interval_ms) m->expected_interval_ms = 1; + } + if (m->lag_ms > STREAM_LATE_MS) bucket->late_blocks++; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "metrics seq=%llu source_delta=%u receive_delta=%llu lag=%llu expected=%llu", + (unsigned long long)seq, source_delta, (unsigned long long)receive_delta, + (unsigned long long)m->lag_ms, (unsigned long long)m->expected_interval_ms); + } + m->last_seq = seq; + m->last_timestamp_ms = timestamp_ms; + m->last_receive_ms = now_ms; + m->initialized = 1; + m->stopped = 0; + return STREAM_OK; +} + +/* Скорость суммируется по объёму и времени, а не усреднением частных rates разных интервалов. */ +int stream_metrics_poll(struct stream_metrics *m, uint64_t now_ms, struct stream_metric_snapshot *snapshot) { + if (!m || !snapshot) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics poll invalid arguments"); return STREAM_ERR_ARGUMENT; } + if (m->initialized && now_ms < m->last_receive_ms) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "metrics poll clock regression now=%llu previous=%llu", + (unsigned long long)now_ms, (unsigned long long)m->last_receive_ms); + return STREAM_ERR_STATE; + } + struct stream_metric_snapshot result = {0}; + if (!m->initialized) { *snapshot = result; return STREAM_WAIT; } + metrics_detect_stop(m, now_ms); + result.lag_ms = m->lag_ms; + result.expected_interval_ms = m->expected_interval_ms; + result.stopped = m->stopped; + if (!m->ended && now_ms - m->last_receive_ms > m->expected_interval_ms) + result.lag_ms = metrics_add(result.lag_ms, now_ms - m->last_receive_ms - m->expected_interval_ms); + uint64_t second = now_ms / 1000; + uint64_t bytes = 0, source_ms = 0, receive_ms = 0; + for (unsigned i = 0; i < STREAM_METRIC_SECONDS; i++) { + const struct stream_metric_bucket *bucket = &m->buckets[i]; + if (!bucket->valid || bucket->second > second || second - bucket->second >= STREAM_METRIC_SECONDS) continue; + result.blocks += bucket->blocks; + result.late_blocks += bucket->late_blocks; + result.stops += bucket->stops; + bytes += bucket->bytes; + source_ms += bucket->source_ms; + receive_ms += bucket->receive_ms; + } + result.source_rate_valid = source_ms != 0; + result.receive_rate_valid = receive_ms != 0; + if (source_ms) result.source_rate_bytes_sec = bytes * 1000 / source_ms; + if (receive_ms) result.receive_rate_bytes_sec = bytes * 1000 / receive_ms; + *snapshot = result; + return STREAM_OK; +} + +/* END останавливает ожидание будущих данных, история качества остаётся доступной. */ +void stream_metrics_end(struct stream_metrics *m) { + if (!m) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics end null context"); return; } + m->ended = 1; + m->stopped = 0; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "metrics ended at seq=%llu", (unsigned long long)m->last_seq); +} diff --git a/src/stream/stream_metrics.h b/src/stream/stream_metrics.h new file mode 100644 index 00000000..a8b35ca1 --- /dev/null +++ b/src/stream/stream_metrics.h @@ -0,0 +1,64 @@ +/* Относительное отставание одного субпотока от одного поставщика, без NTP и сокетов. + * now_ms — монотонные часы получателя; timestamp — неизменяемые 16-битные часы источника. + * Первый блок/разрыв устанавливает новую базу, история проблем остаётся за последние 60 секунд. + * observe/poll вызываются одним uasync-потоком. DUPLICATE не меняет ни часы, ни статистику. */ +#ifndef STREAM_METRICS_H +#define STREAM_METRICS_H + +#include "stream_proto.h" + +#ifdef __cplusplus +extern "C" { +#endif + +#define STREAM_METRIC_SECONDS 60 + +struct stream_metric_bucket { + uint64_t second; + uint64_t blocks; + uint64_t late_blocks; + uint64_t stops; + uint64_t bytes; + uint64_t source_ms; + uint64_t receive_ms; + uint64_t intervals; + int valid; +}; + +struct stream_metrics { + struct stream_metric_bucket buckets[STREAM_METRIC_SECONDS]; + uint64_t last_seq; + uint64_t last_receive_ms; + uint64_t lag_ms; + uint64_t expected_interval_ms; + uint16_t last_timestamp_ms; + int initialized; + int stopped; + int ended; +}; + +struct stream_metric_snapshot { + uint64_t lag_ms; + uint64_t blocks; + uint64_t late_blocks; + uint64_t stops; + uint64_t source_rate_bytes_sec; + uint64_t receive_rate_bytes_sec; + uint64_t expected_interval_ms; + int stopped; + int source_rate_valid; + int receive_rate_valid; +}; + +void stream_metrics_init(struct stream_metrics *metrics); +int stream_metrics_observe(struct stream_metrics *metrics, uint64_t seq, uint16_t timestamp_ms, uint8_t flags, + uint32_t bytes, uint64_t now_ms); +/* Также обнаруживает полную остановку; episode учитывается в момент превышения порога, не на каждом poll. + * До первого блока возвращает WAIT и обнулённый snapshot; при ошибке snapshot остаётся прежним. */ +int stream_metrics_poll(struct stream_metrics *metrics, uint64_t now_ms, struct stream_metric_snapshot *snapshot); +void stream_metrics_end(struct stream_metrics *metrics); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/src/stream/stream_proto.h b/src/stream/stream_proto.h new file mode 100644 index 00000000..d77fb601 --- /dev/null +++ b/src/stream/stream_proto.h @@ -0,0 +1,73 @@ +/* Неизменяемый блок источника: все числа big-endian, подпись охватывает тело до signature. + * Данные ETCP-выдачи (handle/сессия/путь) добавляются снаружи и в подпись не входят. + * seq/subblock берутся из заголовка RS; второго независимого номера блока нет. */ +#ifndef STREAM_PROTO_H +#define STREAM_PROTO_H + +#include +#include + +#include "../../lib/reed_solomon.h" +#include "../transport_layer/secure_channel.h" + +#define STREAM_BLOCK_DOMAIN "stream_block" +#define STREAM_BLOCK_DOMAIN_SIZE 12 +#define STREAM_BLOCK_GROUP_OFF 12 +#define STREAM_BLOCK_SOURCE_OFF 20 +#define STREAM_BLOCK_EPOCH_OFF 28 +#define STREAM_BLOCK_N_OFF 36 +#define STREAM_BLOCK_R_OFF 37 +#define STREAM_BLOCK_SIZE_OFF 38 +#define STREAM_BLOCK_TIMESTAMP_OFF 40 +#define STREAM_BLOCK_LENGTH_OFF 42 +#define STREAM_BLOCK_FLAGS_OFF 46 +#define STREAM_BLOCK_RS_OFF 47 +#define STREAM_BLOCK_SEQ_OFF STREAM_BLOCK_RS_OFF +#define STREAM_BLOCK_SUB_OFF (STREAM_BLOCK_RS_OFF + 8) +#define STREAM_BLOCK_OVERHEAD (STREAM_BLOCK_RS_OFF + SC_SIGN_SIZE) +#define STREAM_BLOCK_DEFAULT_SIZE 1300 +#define STREAM_RS_DEFAULT_SIZE (STREAM_BLOCK_DEFAULT_SIZE - STREAM_BLOCK_OVERHEAD) +#define STREAM_BLOCK_MAX_SIZE (RS_MAX_BLOCK_SIZE + STREAM_BLOCK_OVERHEAD) +#define STREAM_BLOCK_TIME_RESET 0x01 +#define STREAM_BUFFER_BYTES (32u * 1024u * 1024u) +#define STREAM_SOURCE_FLUSH_TB 1000 /* 100 мс, timebase 0.1 мс */ +#define STREAM_LATE_MS 500 +#define STREAM_TIMESTAMP_PERIOD_MS 65536u + +struct stream_descriptor { + uint64_t group_id; + uint64_t source_node_id; + uint64_t source_epoch; + uint16_t block_size; /* Полный размер RS-кодограммы, без stream-метаданных/подписи. */ + uint8_t n; + uint8_t r; +}; + +enum stream_result { + STREAM_OK = 0, + STREAM_DUPLICATE = 1, + STREAM_WAIT = 2, + STREAM_GAP = 3, + STREAM_EXPIRED = 4, + STREAM_ERR_ARGUMENT = -1, + STREAM_ERR_MEMORY = -2, + STREAM_ERR_FORMAT = -3, + STREAM_ERR_SIGNATURE = -4, + STREAM_ERR_CONFLICT = -5, + STREAM_ERR_STATE = -6, + STREAM_ERR_CRYPTO = -7, + STREAM_ERR_FEC = -8 +}; + +/* Короткие примитивы без доступа за границы: размер проверяет вызывающий parser. */ +static inline uint64_t stream_read_be(const uint8_t *data, unsigned bytes) { + uint64_t result = 0; + for (unsigned i = 0; i < bytes; i++) result = (result << 8) | data[i]; + return result; +} + +static inline void stream_write_be(uint8_t *data, uint64_t value, unsigned bytes) { + for (unsigned i = 0; i < bytes; i++) data[bytes - i - 1] = (uint8_t)(value >> (8 * i)); +} + +#endif diff --git a/src/stream/stream_source.c b/src/stream/stream_source.c new file mode 100644 index 00000000..e8bddc48 --- /dev/null +++ b/src/stream/stream_source.c @@ -0,0 +1,181 @@ +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "../../lib/u_async.h" +#include "stream_source.h" + +struct stream_source { + struct UASYNC *ua; + struct stream_descriptor descriptor; + struct stream_encoder *encoder; + struct stream_buffer *buffer; + uint8_t *input; + size_t capacity; + size_t used; + uint64_t next_seq; + uint64_t last_batch_ms; + uint64_t emitted_bytes; + uint64_t report_ms; + void *timer; + stream_source_ready_fn ready; + void *arg; + int busy; + int has_batch; + int failed; + int ended; +}; + +/* Ошибка не скрывает уже принятый префикс: прекращаем таймер и дальнейшие write. */ +static int source_fail(struct stream_source *source, int result) { + source->failed = 1; + if (source->timer) { uasync_cancel_timeout(source->ua, source->timer); source->timer = NULL; } + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source failed source=%llx seq=%llu pending=%zu result=%d", + (unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq, source->used, result); + return result; +} + +/* Сформировать блоки и сохранить их до уведомления потребителей; busy установлен внешней операцией. */ +static int source_flush(struct stream_source *source) { + if (!source->used) return STREAM_OK; + if (source->timer) { uasync_cancel_timeout(source->ua, source->timer); source->timer = NULL; } + uint64_t now_ms = get_time_tb() / 10; + if (source->next_seq == UINT64_MAX || (source->has_batch && now_ms < source->last_batch_ms)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source sequence/clock invalid seq=%llu now=%llu previous=%llu", + (unsigned long long)source->next_seq, (unsigned long long)now_ms, (unsigned long long)source->last_batch_ms); + return source_fail(source, STREAM_ERR_STATE); + } + uint8_t flags = source->has_batch && now_ms - source->last_batch_ms >= STREAM_TIMESTAMP_PERIOD_MS ? STREAM_BLOCK_TIME_RESET : 0; + const uint8_t *blocks[RS_MAX_BLOCKS]; + int result = stream_encoder_encode(source->encoder, source->next_seq, (uint16_t)now_ms, flags, source->input, source->used, blocks); + if (result != STREAM_OK) return source_fail(source, result); + for (unsigned i = 0; i < source->descriptor.n; i++) { + result = stream_buffer_put(source->buffer, blocks[i], stream_block_size(&source->descriptor)); + if (result != STREAM_OK) return source_fail(source, result < 0 ? result : STREAM_ERR_STATE); + } + uint64_t seq = source->next_seq++; + source->emitted_bytes += source->used; + source->used = 0; + source->has_batch = 1; + source->last_batch_ms = now_ms; + if (!source->report_ms || now_ms - source->report_ms >= 1000) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source progress source=%llx seq=%llu bytes=%llu timestamp=%u flags=%u", + (unsigned long long)source->descriptor.source_node_id, (unsigned long long)seq, + (unsigned long long)source->emitted_bytes, (uint16_t)now_ms, flags); + source->report_ms = now_ms; + } + if (source->ready) source->ready(source, seq, source->arg); + return STREAM_OK; +} + +/* Срабатывание единственного таймера неполной пачки; handle больше не принадлежит таймерному heap. */ +static void source_flush_timer(void *arg) { + struct stream_source *source = arg; + source->timer = NULL; + source->busy = 1; + int result = source_flush(source); + source->busy = 0; + if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source timed flush failed result=%d", result); +} + +/* Создать только реальную локальную публикацию; память удалённых анонсов сюда не попадает. */ +int stream_source_create(struct stream_source **out, struct UASYNC *ua, const struct stream_descriptor *d, + const uint8_t private_key[32], stream_source_ready_fn ready, void *arg) { + if (!out || *out || !ua || !private_key || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source create invalid arguments out=%p ua=%p", (void *)out, (void *)ua); + return STREAM_ERR_ARGUMENT; + } + struct stream_source *source = u_calloc(1, sizeof(*source)); + if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source context allocation failed"); return STREAM_ERR_MEMORY; } + source->ua = ua; + source->descriptor = *d; + source->ready = ready; + source->arg = arg; + source->capacity = stream_input_size(d); + source->input = u_malloc((uint32_t)source->capacity); + int result = source->input ? stream_encoder_create(&source->encoder, d, private_key) : STREAM_ERR_MEMORY; + if (result == STREAM_OK) + result = stream_buffer_create(&source->buffer, d, stream_encoder_public_key(source->encoder), STREAM_BUFFER_BYTES); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source initialization failed source=%llx result=%d", (unsigned long long)d->source_node_id, result); + stream_source_destroy(source); + return result; + } + *out = source; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source created group=%llx source=%llx epoch=%llx input=%zu flush=100ms", + (unsigned long long)d->group_id, (unsigned long long)d->source_node_id, + (unsigned long long)d->source_epoch, source->capacity); + return STREAM_OK; +} + +void stream_source_destroy(struct stream_source *source) { + if (!source) return; + if (source->busy) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source destroy inside callback rejected"); return; } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source destroy source=%llx batches=%llu bytes=%llu pending=%zu ended=%d failed=%d", + (unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq, + (unsigned long long)source->emitted_bytes, source->used, source->ended, source->failed); + if (source->timer) uasync_cancel_timeout(source->ua, source->timer); + stream_buffer_destroy(source->buffer); + stream_encoder_destroy(source->encoder); + u_free(source->input); + u_free(source); +} + +size_t stream_source_input_capacity(const struct stream_source *source) { + if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source capacity null context"); return 0; } + return source->capacity; +} + +struct stream_buffer *stream_source_buffer(struct stream_source *source) { + if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source buffer null context"); return NULL; } + return source->buffer; +} + +/* Принимать ограниченную порцию, дополнять старую пачку и заводить таймер только на новый остаток. */ +int stream_source_write(struct stream_source *source, const uint8_t *data, size_t length, size_t *accepted) { + if (accepted) *accepted = 0; + if (!source || !accepted || (length && !data) || length > source->capacity) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source write invalid arguments length=%zu", length); + return STREAM_ERR_ARGUMENT; + } + if (source->busy || source->failed || source->ended) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source write state rejected busy=%d failed=%d ended=%d", source->busy, source->failed, source->ended); + return STREAM_ERR_STATE; + } + if (!length) return STREAM_OK; + source->busy = 1; + int result = STREAM_OK; + while (*accepted < length) { + size_t take = source->capacity - source->used; + if (take > length - *accepted) take = length - *accepted; + memcpy(source->input + source->used, data + *accepted, take); + source->used += take; + *accepted += take; + if (source->used == source->capacity && (result = source_flush(source)) != STREAM_OK) break; + } + if (result == STREAM_OK && source->used && !source->timer) { + source->timer = uasync_set_timeout(source->ua, STREAM_SOURCE_FLUSH_TB, source, source_flush_timer, "stream_source_flush"); + if (!source->timer) result = source_fail(source, STREAM_ERR_MEMORY); + } + source->busy = 0; + return result; +} + +/* Finish отменяет таймер и публикует остаток сразу; дальнейшие байты требуют нового source_epoch. */ +int stream_source_finish(struct stream_source *source) { + if (!source || source->busy || source->failed) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source finish invalid state present=%d", source != NULL); + return source ? STREAM_ERR_STATE : STREAM_ERR_ARGUMENT; + } + if (source->ended) return STREAM_OK; + source->busy = 1; + int result = source_flush(source); + source->busy = 0; + if (result == STREAM_OK) { + source->ended = 1; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source finished source=%llx batches=%llu bytes=%llu", + (unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq, + (unsigned long long)source->emitted_bytes); + } + return result; +} diff --git a/src/stream/stream_source.h b/src/stream/stream_source.h new file mode 100644 index 00000000..364fcf71 --- /dev/null +++ b/src/stream/stream_source.h @@ -0,0 +1,36 @@ +/* Источник непрозрачных байтов: одна подписанная FEC-пачка за ограниченный write, + * неполная пачка сбрасывается таймером 100 мс, все блоки сначала сохраняются в кольце. + * Один uasync-поток; источник сам владеет encoder/кольцом/таймером. NCD здесь ещё нет. + * ready вызывается после помещения всей пачки и может читать кольцо; write/finish/destroy + * из ready запрещены. destroy выполняется вне callbacks источника. */ +#ifndef STREAM_SOURCE_H +#define STREAM_SOURCE_H + +#include "stream_buffer.h" + +#ifdef __cplusplus +extern "C" { +#endif + +struct UASYNC; +struct stream_source; +typedef void (*stream_source_ready_fn)(struct stream_source *source, uint64_t seq, void *arg); + +int stream_source_create(struct stream_source **out, struct UASYNC *ua, const struct stream_descriptor *descriptor, + const uint8_t private_key[SC_PRIVKEY_SIZE], stream_source_ready_fn ready, void *arg); +void stream_source_destroy(struct stream_source *source); +size_t stream_source_input_capacity(const struct stream_source *source); +/* Только заимствованное чтение/decode; вставлять блоки/освобождать это кольцо должен сам источник. */ +struct stream_buffer *stream_source_buffer(struct stream_source *source); + +/* length <= input_capacity: вызов кодирует не более одной пачки, не захватывая uasync большим внешним буфером. + * accepted получает принятую часть даже при ошибке; нулевой write допустим и не создаёт пустую пачку. + * Ошибка публикации/таймера делает источник failed; его уничтожают, не повторяют принятую часть как новую трансляцию. */ +int stream_source_write(struct stream_source *source, const uint8_t *data, size_t length, size_t *accepted); +/* Сбросить остаток и остановить приём; повторный finish допустим, кольцо доступно до destroy. */ +int stream_source_finish(struct stream_source *source); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/tests/Makefile.am b/tests/Makefile.am index d2cd0dff..c44c39d4 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -93,6 +93,7 @@ endif # Короткие интеграционные проверки (до 1с). check_PROGRAMS += \ + test_stream_core \ test_u_async_comprehensive \ test_nat_detection \ test_audio_compressor \ @@ -444,6 +445,10 @@ test_reed_solomon_SOURCES = test_reed_solomon.c test_reed_solomon_CFLAGS = -I$(top_srcdir)/lib test_reed_solomon_LDADD = $(COMMON_LIBS) +test_stream_core_SOURCES = test_stream_core.c $(test_clock_sources) +test_stream_core_LDADD = $(LIBUTUN) $(COMMON_LIBS) +test_stream_core_LDFLAGS = $(test_clock_ldflags) + test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_intensive_memory_pool_LDADD = $(COMMON_LIBS) diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c new file mode 100644 index 00000000..aa085049 --- /dev/null +++ b/tests/test_stream_core.c @@ -0,0 +1,326 @@ +/* Подпись источника, RS-потери, кольцо, циклическое время и реальные uasync-таймеры источника. */ +#include +#include +#include + +#include "../src/stream/stream_source.h" +#include "../src/stream/stream_metrics.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "test_clock.h" + +#define CHECK(expr) do { if (!(expr)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_core failed line=%d: %s", __LINE__, #expr); exit(1); \ +} } while (0) + +static const struct stream_descriptor descriptor = { + .group_id = UINT64_C(0x0123456789abcdef), .source_node_id = UINT64_C(0xfedcba9876543210), + .source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2 +}; +static uint8_t private_key[32]; + +/* Воспроизводимые данные, не совпадающие с zero-padding и проверочными кодограммами. */ +static void pattern(uint8_t *data, size_t length) { + uint32_t state = 0x12345678; + for (size_t i = 0; i < length; i++) { + state ^= state << 13; state ^= state >> 17; state ^= state << 5; + data[i] = (uint8_t)state; + } +} + +/* Подмена любого класса неизменяемых полей отвергается, view и входной блок сохраняются. */ +static void test_signed_blocks(void) { + struct stream_encoder *encoder = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + CHECK(stream_block_size(&descriptor) == 1300 && stream_input_size(&descriptor) == 7080); + uint8_t input[7080]; pattern(input, sizeof(input)); + const uint8_t *blocks[RS_MAX_BLOCKS]; + uint64_t seq = UINT64_C(0x1122334455667788); + CHECK(stream_encoder_encode(encoder, seq, 65530, STREAM_BLOCK_TIME_RESET, input, sizeof(input), blocks) == STREAM_OK); + const uint8_t *public_key = stream_encoder_public_key(encoder); + for (unsigned i = 0; i < descriptor.n; i++) { + struct stream_block_view view; + CHECK(stream_block_verify(&descriptor, public_key, blocks[i], 1300, &view) == STREAM_OK); + CHECK(view.seq == seq && view.subblock == i && view.timestamp_ms == 65530 && view.data_length == sizeof(input)); + CHECK(view.flags == STREAM_BLOCK_TIME_RESET && view.rs == blocks[i] + STREAM_BLOCK_RS_OFF); + CHECK(!memcmp(blocks[i], "stream_block", 12)); + CHECK(!memcmp(blocks[i] + STREAM_BLOCK_GROUP_OFF, "\x01\x23\x45\x67\x89\xab\xcd\xef", 8)); + CHECK(!memcmp(blocks[i] + STREAM_BLOCK_SEQ_OFF, "\x11\x22\x33\x44\x55\x66\x77\x88", 8)); + CHECK(sc_ed25519_verify(public_key, blocks[i], 1300 - 64, blocks[i] + 1300 - 64) == SC_OK); + } + const unsigned changed[] = {0, 12, 20, 28, 36, 37, 38, 40, 45, 46, 47, 55, 56, 1235, 1299}; + uint8_t wire[STREAM_BLOCK_MAX_SIZE]; + for (unsigned i = 0; i < sizeof(changed) / sizeof(changed[0]); i++) { + memcpy(wire, blocks[0], 1300); wire[changed[i]] ^= 1; + struct stream_block_view view; memset(&view, 0xa5, sizeof(view)); + uint8_t before[sizeof(view)]; memcpy(before, &view, sizeof(view)); + CHECK(stream_block_verify(&descriptor, public_key, wire, 1300, &view) < 0); + CHECK(!memcmp(before, &view, sizeof(view))); + } + struct stream_block_view view; + CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 0, &view) == STREAM_ERR_FORMAT); + CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1299, &view) == STREAM_ERR_FORMAT); + CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1301, &view) == STREAM_ERR_FORMAT); + uint8_t wrong_key[32]; memcpy(wrong_key, public_key, 32); wrong_key[0] ^= 1; + CHECK(stream_block_verify(&descriptor, wrong_key, blocks[0], 1300, &view) == STREAM_ERR_SIGNATURE); + CHECK(stream_encoder_encode(encoder, UINT64_MAX, 0, 0, input, 1, blocks) == STREAM_ERR_ARGUMENT); + CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, 0, blocks) == STREAM_ERR_ARGUMENT); + CHECK(stream_encoder_encode(encoder, 1, 0, 2, input, 1, blocks) == STREAM_ERR_ARGUMENT); + struct stream_descriptor invalid = descriptor; invalid.n = 0; + CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); + invalid = descriptor; invalid.r = invalid.n; + CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); + invalid = descriptor; invalid.block_size = 9; + CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); + invalid = descriptor; invalid.source_epoch = 0; + CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); + stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes"); +} + +/* Любая пара потерь при 6+2 восстанавливается из подписанных блоков, отсутствующие блоки не регенерируются. */ +static void test_fec_and_padding(void) { + struct stream_encoder *encoder = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + uint8_t input[7080], output[7080]; pattern(input, sizeof(input)); + const uint8_t *blocks[RS_MAX_BLOCKS]; + CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks) == STREAM_OK); + for (unsigned first = 0; first < 8; first++) { + for (unsigned second = first + 1; second < 8; second++) { + struct stream_buffer *buffer = NULL; + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK); + for (unsigned i = 0; i < 8; i++) + if (i != first && i != second) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); + size_t length = 0; + memset(output, 0xcc, sizeof(output)); + size_t allocations = u_get_allocated_count(); + CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_OK); + CHECK(u_get_allocated_count() == allocations && length == sizeof(input) && !memcmp(input, output, length)); + const uint8_t *missing = NULL; size_t missing_length = 0; uint64_t actual = 100; + CHECK(stream_buffer_read(buffer, 0, (uint8_t)first, &missing, &missing_length, &actual) == STREAM_GAP); + CHECK(!missing && missing_length == 0 && actual == 100); + struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK); + CHECK(stats.blocks == 6 && stats.stored_bytes == 6 * 1300); + stream_buffer_destroy(buffer); + } + } + struct stream_buffer *buffer = NULL; + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK); + for (unsigned i = 0; i < 5; i++) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); + memset(output, 0xcc, sizeof(output)); size_t length = 123; + CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_WAIT); + CHECK(length == 123); + for (unsigned i = 0; i < sizeof(output); i++) CHECK(output[i] == 0xcc); + CHECK(stream_buffer_put(buffer, blocks[5], 1300) == STREAM_OK); + CHECK(stream_buffer_decode(buffer, 0, output, 1, &length) == STREAM_ERR_ARGUMENT); + stream_buffer_destroy(buffer); + stream_encoder_destroy(encoder); + + const struct { uint8_t n, r; uint16_t size; } variants[] = {{1, 0, 1189}, {3, 1, 20}, {16, 5, 4096}, {255, 254, 10}}; + for (unsigned v = 0; v < sizeof(variants) / sizeof(variants[0]); v++) { + struct stream_descriptor d = descriptor; d.n = variants[v].n; d.r = variants[v].r; d.block_size = variants[v].size; + encoder = NULL; buffer = NULL; + CHECK(stream_encoder_create(&encoder, &d, private_key) == STREAM_OK); + size_t size = stream_block_size(&d), wanted = stream_input_size(&d) > 17 ? 17 : stream_input_size(&d); + CHECK(stream_encoder_encode(encoder, 37, 0, 0, input, wanted, blocks) == STREAM_OK); + CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), d.n * size) == STREAM_OK); + for (unsigned i = d.r; i < d.n; i++) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK); + uint8_t partial[17]; length = 0; + CHECK(stream_buffer_decode(buffer, 37, partial, wanted, &length) == STREAM_OK); + CHECK(length == wanted && !memcmp(partial, input, wanted)); + stream_buffer_destroy(buffer); stream_encoder_destroy(encoder); + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] FEC: all 28 double losses, insufficient set, partial data, n=1..255 and r=0"); +} + +/* Вставлять подписанную пачку в тестовое кольцо до повторного использования буферов encoder. */ +static void put_batch(struct stream_encoder *encoder, struct stream_buffer *buffer, uint64_t seq, unsigned mask) { + uint8_t input[37]; pattern(input, sizeof(input)); + const uint8_t *blocks[RS_MAX_BLOCKS]; + CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, sizeof(input), blocks) == STREAM_OK); + for (unsigned i = 0; i < 8; i++) if (mask & (1u << i)) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); +} + +/* Общее кольцо хранит реальные блоки, принимает порядок 10/12/11 и вытесняет целую минимальную пачку. */ +static void test_ring_and_conflicts(void) { + struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK); + const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = 100; + CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_WAIT); + put_batch(encoder, buffer, 10, 255); put_batch(encoder, buffer, 12, 255); + CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_OK); + CHECK(stream_buffer_put(buffer, wire, length) == STREAM_DUPLICATE); + uint8_t bad[1300]; memcpy(bad, wire, length); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1; + CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_SIGNATURE); + CHECK(sc_ed25519_sign(private_key, bad, sizeof(bad) - 64, bad + sizeof(bad) - 64) == SC_OK); + CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_CONFLICT); + put_batch(encoder, buffer, 11, 1); + CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_EXPIRED); + CHECK(stream_buffer_read(buffer, 11, 1, &wire, &length, &actual) == STREAM_GAP); + CHECK(stream_buffer_read(buffer, 13, 0, &wire, &length, &actual) == STREAM_WAIT); + CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == 12); + uint64_t first = 0, last = 0; + CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 11 && last == 12); + struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK); + CHECK(stats.blocks == 9 && stats.batches == 2 && stats.stored_bytes == 9 * 1300 && stats.expired_before == 11); + const uint8_t *blocks[RS_MAX_BLOCKS]; uint8_t input[37]; pattern(input, sizeof(input)); + CHECK(stream_encoder_encode(encoder, 10, 10, 0, input, sizeof(input), blocks) == STREAM_OK); + CHECK(stream_buffer_put(buffer, blocks[0], 1300) == STREAM_EXPIRED); + put_batch(encoder, buffer, 13, 255); + CHECK(stream_buffer_read(buffer, 11, 0, &wire, &length, &actual) == STREAM_EXPIRED); + put_batch(encoder, buffer, UINT64_MAX - 1, 1); + CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == UINT64_MAX - 1); + CHECK(stream_buffer_read(buffer, UINT64_MAX, 7, &wire, &length, &actual) == STREAM_OK && actual == 13); + CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.stored_bytes <= stats.capacity_bytes); + stream_buffer_destroy(buffer); + + buffer = NULL; + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK); + for (unsigned seq = 0; seq < 16; seq++) put_batch(encoder, buffer, seq, 1); + CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.blocks == 16 && stats.batches == 16); + CHECK(stats.stored_bytes == stats.capacity_bytes); /* Ни один из семи пустых субпотоков не резервирует место. */ + put_batch(encoder, buffer, 16, 1); + CHECK(stream_buffer_read(buffer, 0, 0, &wire, &length, &actual) == STREAM_EXPIRED); + CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 1 && last == 16); + stream_buffer_destroy(buffer); stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] ring: shared data budget, sparse batches, out-of-order indices, whole-batch eviction and conflicts"); +} + +/* Подлинное время источника разворачивается через ноль; постоянный lag не считается полной остановкой. */ +static void test_relative_metrics(void) { + struct stream_metrics m; struct stream_metric_snapshot snapshot; + stream_metrics_init(&m); + CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_WAIT); + CHECK(stream_metrics_observe(&m, 100, 65520, 0, 1300, 1000) == STREAM_OK); + CHECK(stream_metrics_observe(&m, 101, 4, 0, 1300, 1020) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 1020, &snapshot) == STREAM_OK); + CHECK(snapshot.lag_ms == 0 && snapshot.source_rate_bytes_sec == 65000 && snapshot.expected_interval_ms == 20); + CHECK(snapshot.receive_rate_bytes_sec == 65000 && snapshot.blocks == 2 && !snapshot.stopped); + CHECK(stream_metrics_observe(&m, 102, 24, 0, 1300, 1320) == STREAM_OK); + CHECK(stream_metrics_observe(&m, 103, 44, 0, 1300, 1620) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 1620, &snapshot) == STREAM_OK); + CHECK(snapshot.lag_ms == 560 && snapshot.late_blocks == 1 && snapshot.stops == 0 && !snapshot.stopped); + uint64_t seq = 103, now = 1620; uint16_t timestamp = 44; + for (unsigned i = 0; i < 30; i++) { + seq++; now++; timestamp = (uint16_t)(timestamp + 20); + CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK); + } + CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && snapshot.lag_ms == 0); + CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now + 300) == STREAM_DUPLICATE); + CHECK(stream_metrics_poll(&m, now + 521, &snapshot) == STREAM_OK); + CHECK(snapshot.stopped && snapshot.stops == 1 && snapshot.lag_ms == 501); + CHECK(stream_metrics_poll(&m, now + 800, &snapshot) == STREAM_OK && snapshot.stops == 1); + seq++; timestamp = (uint16_t)(timestamp + 20); now += 800; + CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK); + CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && !snapshot.stopped && snapshot.stops == 1); + CHECK(stream_metrics_observe(&m, seq + 1, timestamp, 0, 1300, now - 1) == STREAM_ERR_STATE); + stream_metrics_end(&m); + CHECK(stream_metrics_poll(&m, now + 70000, &snapshot) == STREAM_OK); + CHECK(!snapshot.blocks && !snapshot.late_blocks && !snapshot.stops && !snapshot.stopped && !snapshot.source_rate_valid); + stream_metrics_init(&m); + CHECK(stream_metrics_observe(&m, 0, 0, 0, 1300, 0) == STREAM_OK); + CHECK(stream_metrics_observe(&m, 1, 0, 0, 1300, 0) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_OK && !snapshot.source_rate_valid); + CHECK(stream_metrics_observe(&m, 2, 1, 0, 1300, 1) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 1, &snapshot) == STREAM_OK && snapshot.source_rate_bytes_sec == 2600000); + CHECK(stream_metrics_observe(&m, 3, 9, STREAM_BLOCK_TIME_RESET, 1300, 2) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 2, &snapshot) == STREAM_OK && snapshot.lag_ms == 0); + CHECK(stream_metrics_observe(&m, 5, 10, 0, 1300, 1000) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 1000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0); + CHECK(stream_metrics_observe(&m, 6, 11, 0, 1300, 71000) == STREAM_OK); + CHECK(stream_metrics_poll(&m, 71000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0 && !snapshot.stopped); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] metrics: timestamp wrap, zero deltas, lag/catchup, complete stall, one episode and window expiry"); +} + +struct source_events { + unsigned count; + uint64_t seq; + uint64_t time_tb; + uint16_t timestamp_ms; + uint8_t flags; +}; + +/* Callback видит уже всю пачку, а не частично опубликованные подписи. */ +static void source_ready(struct stream_source *source, uint64_t seq, void *arg) { + struct source_events *events = arg; + events->count++; events->seq = seq; events->time_tb = get_time_tb(); + struct stream_buffer_stats stats; + CHECK(stream_buffer_get_stats(stream_source_buffer(source), &stats) == STREAM_OK); + CHECK(stats.blocks == events->count * 8 && stats.highest_seq == seq); + const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = UINT64_MAX; + CHECK(stream_buffer_read(stream_source_buffer(source), seq, 0, &wire, &length, &actual) == STREAM_OK); + CHECK(actual == seq && length == STREAM_BLOCK_DEFAULT_SIZE); + events->timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2); + events->flags = wire[STREAM_BLOCK_FLAGS_OFF]; + for (unsigned sub = 1; sub < 8; sub++) { + CHECK(stream_buffer_read(stream_source_buffer(source), seq, (uint8_t)sub, &wire, &length, &actual) == STREAM_OK); + CHECK(stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2) == events->timestamp_ms); + CHECK(wire[STREAM_BLOCK_FLAGS_OFF] == events->flags); + } +} + +/* Настоящий uasync timer: не сбрасывает до 100 мс, полный блок отменяет timer, destroy отменяет остаток. */ +static void test_source_timer(void) { + test_clock_init(50); + struct UASYNC *ua = uasync_create(); CHECK(ua != NULL); + struct stream_source *source = NULL; struct source_events events = {0}; + CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK); + uint8_t input[7080], output[7080]; pattern(input, sizeof(input)); + CHECK(stream_source_input_capacity(source) == sizeof(input)); + size_t accepted = 999; + CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0); + uint64_t start = get_time_tb(); + CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK && accepted == 17); + test_clock_run_for(ua, 500); + CHECK(events.count == 0); + test_clock_next_timer(ua); + CHECK(events.count == 1 && events.seq == 0 && events.time_tb - start >= STREAM_SOURCE_FLUSH_TB); + CHECK((uint16_t)(events.timestamp_ms - (uint16_t)(start / 10)) >= 100 && !events.flags); + size_t length = 0; + CHECK(stream_buffer_decode(stream_source_buffer(source), 0, output, 17, &length) == STREAM_OK); + CHECK(length == 17 && !memcmp(input, output, length)); + CHECK(stream_source_write(source, input, sizeof(input), &accepted) == STREAM_OK && accepted == sizeof(input)); + CHECK(events.count == 2 && events.seq == 1); + test_clock_run_for(ua, 1100); + CHECK(events.count == 2); /* Нет лишней пустой пачки. */ + CHECK(stream_source_write(source, input, 13, &accepted) == STREAM_OK && accepted == 13); + CHECK(stream_source_write(source, input + 13, sizeof(input) - 13, &accepted) == STREAM_OK); + CHECK(events.count == 3 && events.seq == 2); + test_clock_run_for(ua, 1100); + CHECK(events.count == 3); /* Timer неполной пачки отменён заполнением. */ + CHECK(stream_buffer_decode(stream_source_buffer(source), 2, output, sizeof(output), &length) == STREAM_OK); + CHECK(length == sizeof(input) && !memcmp(input, output, length)); + test_clock_run_for(ua, STREAM_TIMESTAMP_PERIOD_MS * 10); + CHECK(stream_source_write(source, input, 23, &accepted) == STREAM_OK); + CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4); + CHECK(events.flags == STREAM_BLOCK_TIME_RESET); /* Источник явно пометил неоднозначное число оборотов. */ + CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4); + CHECK(stream_source_write(source, input, 1, &accepted) == STREAM_ERR_STATE && accepted == 0); + stream_source_destroy(source); + source = NULL; events.count = 0; + CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK); + CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK); + stream_source_destroy(source); + test_clock_run_for(ua, 1100); + CHECK(events.count == 0); + uasync_destroy(ua, 1); test_clock_finish(); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source: 100ms timer, immediate full batch, padding, finish and cancellation on destroy"); +} + +int main(void) { + debug_config_init(); + debug_set_level(DEBUG_LEVEL_ERROR); + debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_DISABLED); + CHECK(get_category_by_name("stream") == DEBUG_CATEGORY_STREAM); + for (unsigned i = 0; i < sizeof(private_key); i++) private_key[i] = (uint8_t)(i * 37 + 11); + size_t allocations = u_get_allocated_count(); + test_signed_blocks(); CHECK(u_get_allocated_count() == allocations); + test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations); + test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations); + test_relative_metrics(); CHECK(u_get_allocated_count() == allocations); + test_source_timer(); CHECK(u_get_allocated_count() == allocations); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream core: all checks, no tracked allocations remain"); + return 0; +}