utun2/doc/stream.txt

Модуль стриминга. Техническое задание, этап 1.

1. Цель и принятые решения

Модуль размещается в src/stream. Узлы получают поток данных от источника через других подписчиков группы,
передают его своим подписчикам и постоянно ищут более качественные пути доставки отдельных субпотоков.
Этап 1 использует CHAT-группы с таблицей мемберов и member_sync/merkle_sync.

Принятые решения:
- Транспорт — существующий ETCP с быстрыми переповторами, мультилинком, шифрованием и управлением нагрузкой.
- Параметры FEC задаёт источник; 6+2 — значение по умолчанию, а не ограничение протокола.
- Источник подписывает каждый блок после разбиения на субпотоки, включая проверочные блоки.
- В источник добавляется циклический timestamp uint16_t в миллисекундах; он сохраняется при всей ретрансляции.
  Задержка оценивается относительно предыдущих данных по дельтам timestamp и времени прихода, включая полную остановку.
- Передаём произвольные данные как есть, без обработки форматов видео, кадров и ключевых точек.
- Размер одной кодограммы после разбиения — около 1300 байт; источник сбрасывает неполную пачку через 100 мс.
- На каждый локальный экземпляр трансляции — кольцевой буфер 32 МБ для всех её субпотоков вместе.
  Вся выдача идёт из этого буфера. Вытесненные данные недоступны: не успел получить — опоздал.
- Отдельных дозапросов диапазонов нет. Единственный способ получения — запросить субпоток с индекса или с конца.
- Обратное подключение — NCD + REVERSE_REQ стриминга через маршрутизатор группы.
- Статус источника и список подписок — опции самого мембера, распространяемые через Merkle.
  Удалённая подписка актуальна только пока узел онлайн в BGP данной группы.
- Экземпляры определяются локальными подписками; узлы без подписки не принимают и не ретранслируют поток.
  Активный источник также имеет собственный экземпляр и подписку на свою трансляцию.
- Подтверждение перегрузки — 500 мс. Число отключаемых выдач определяется ростом очереди и rate отдельных выдач.
- При запуске параллельно пробуем до 32 совместимых кандидатов; далее одна проба в секунду,
  новый полный цикл не начинается раньше чем через 30 секунд от начала предыдущего.

Обратная совместимость с прежними форматами не требуется.

2. Трансляция, подписки и владение ресурсами

Трансляция определяется ключом (group_id, source_node_id, source_epoch).
source_epoch — новый случайный ненулевой 64-битный идентификатор при каждом запуске источника.
В этапе 1 у одного источника в одной группе активна одна трансляция. Старые и новые запуски различаются по эпохе.
Пакеты и уведомления старого запуска не могут менять состояние нового.

Локальная подписка на трансляцию создаёт один экземпляр, общий для получения и ретрансляции.
Подписка на трансляцию отличается от сетевой выдачи конкретного субпотока по конкретному 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.
Зафиксированный формат:
    stream_source = "<source_epoch:16 lowercase hex>:<n:decimal>:<r:decimal>:<block_size:decimal>"
    stream_subscriptions = "<source_node_id:16 lowercase hex>:<source_epoch:16 lowercase hex>,..."
Отсутствующая/пустая опция означает отсутствие активности. Подписки отсортированы по source_node_id, затем эпохе;
нулевые IDs/эпохи, дубликаты, неверные разделители, ведущие нули decimal и переполнение отвергаются.
Активная публикация требует записи собственной трансляции в stream_subscriptions.
Максимум элементов списка — MS_USERINFO_MAX/34; фактически доступное число дополнительно ограничено
размером всего JSON с именем, аватаркой и прочими сохранёнными опциями. Две stream-опции меняются атомарно.
Источник включает свою трансляцию в собственные подписки.
Список относится к трансляциям, а не к сетевым 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.

[+] Включение в жизненный цикл чат-сервиса — 11.10.2026:
    stream.c/h — контекст в UTUN_INSTANCE, локальные handles публикации/подписки, общий экземпляр и буфер.
    DATA/GAP/END/CLOSED доставляются отложенно через один callback экземпляра ядра;
    callback может закрывать handles, заимствованные данные живут до его возврата.
    stream_member.c/h — канонические опции, проверка подписанного профиля/ключа и членства,
    online-кандидаты только из живых BGP-путей этой CHAT-группы; мобильный источник допускается, relay исключается.
    chat_core_set_my_profile_options — атомарная замена опций с сохранением имени, аватарки и остальных userinfo.
    При старте снимаются оставшиеся после аварии собственные опции; смена эпохи/ключа или утрата членства отменяют подписку.
    Удаление группы и остановка CHAT отменяют таймеры/события, отсоединяют handles и освобождают буферы;
    detached handle остаётся во владении вызывающего до stream_close, общий UTUN/транспорт сохраняется.
    test_stream_service: четыре живых ядра, Merkle-обмен через ETCP, две группы, общий буфер двух владельцев,
    BGP withdraw с сохранённой опцией, смена эпохи, потеря своего членства, удаление канала, restart и отмена таймера.
    Проверены rollback размера/SQLite, FEC-доставка непрозрачных байтов и закрытие из DATA callback, DATA перед END.
    Тест прошёл, счётчик живых аллокаций после destroy вернулся к нулю. Лог: /tmp/stream-service-test-final.log.
    Найденная тестом утечка существующих BGP-путей исправлена: записи освобождаются до контейнера node_paths.
    Общий ./check.sh: 144 passed, 0 failed, 0 skipped; proxy/full, burst и load — PASS.
    Лог общего прогона: /tmp/stream-service-check.log (107 с).

[+] Сетевая выдача — 11.10.2026:
    stream_wire.c/h — big-endian формат, raw ETCP_ID_STREAM=0x07, routed ETCP_RT_ID_STREAM=0x39,
    INFO/ACCEPT с границами окна и путём, 64-битные эпохи/handles, отдельная подпись источника на END.
    stream_network.c/h — собственный NCD owner пира, общий для его потоков/групп; ожидание его UP,
    прямые SUBSCRIBE/ACCEPT/REJECT/DATA/GAP/RENEW/UNSUBSCRIBE/CLOSED/END, routed REVERSE_REQ/CLOSED.
    Новые handles назначаются при постановке SUBSCRIBE в общую FIFO; watermark защищает от закрытых/старых запросов.
    Первый проверенный DATA с конца устанавливает общий N; прочие latest-выдачи ждут его и запрашивают конкретный N.
    Повторный stream_receive_restart отменяет upstream старой попытки, оставляя кольцо/downstream;
    старые receive handles отсоединяются и требуют close. Уже сохранённый первый блок также устанавливает новый старт.
    DATA копируется из кольца только перед отправкой; пути сохраняются по каждому блоку и проверяются на LOOP.
    Все выбранные upstream, прошедшие недостающую пачку, позволяют потребителю выдать явный GAP и продолжить.
    END подписывает domain/ключ запуска/последний индекс; нельзя завершить выдачу до прохождения её последнего блока/пропуска.
    Период RENEW=2с, аренда=6с, первая попытка=5с; reverse не чаще 1/с в этом сроке.
    DOWN/REINIT, потеря членства и закрытие локальной подписки отменяют относящиеся выдачи;
    закрытие одной группы/выдачи сохраняет чужое владение транспортом и другие группы на том же пиру.
    test_stream_service расширен настоящей цепочкой источник -> relay -> получатель поверх ETCP,
    общим индексом, новым bootstrap, FEC 6+2 после снятия двух субпотоков, GAP, LOOP, повторами SUBSCRIBE/reverse,
    подписанным END, продлениями/таймаутом аренды и асинхронным REINIT.
    Две группы с FEC 6+2 и 1+0 используют один ETCP: истечение выдач первой сохраняет DATA второй.
    test_stream_core проверяет wire: все 64 бита, усечения, версии, пути/циклы и размеры.
    Оба теста прошли под ASAN/LeakSanitizer с detect_leaks=1 и halt_on_error=1, без ошибок и остатка живых аллокаций.
    Логи: /tmp/stream-network-asan-core.log и /tmp/stream-network-asan-service.log.
    Обычная сборка восстановлена: ./build.sh --no-gui -j4 прошёл.
    Общий ./check.sh: 144 passed, 0 failed, 0 skipped; proxy/full (9/9), burst и load — PASS.
    Лог: /tmp/stream-network-check.log (88 с).

[ ] Планировщик отправки и разгрузка:
    одна очередь на пира, backpressure, ограничение памяти/возраста, A/D/X/Q,
    подтверждение 500 мс, сокращение по g/rates и двухсекундный срок исполнения OVERLOAD.

[ ] Оптимизация сети и интеграционные проверки:
    пути/защита от циклов, стартовые 32 пробы, дальнейший темп 1/с и полный цикл не чаще 30 с,
    постепенный перенос с перекрытием, failover, несколько источников и независимое закрытие на общем ETCP.

Сетевая выдача подключена: stream_receive_open выбирает явного пира/субпоток/позицию,
stream_receive_get_info сообщает handle/прогресс/причину, stream_receive_close снимает только свою выдачу.
Локальная подписка по-прежнему владеет кольцом; receive handle её не заменяет и закрывается отдельно.
Прямой dispatcher в старой части теста сохранён для изолированной проверки FEC и закрытия из callback;
новая часть проверяет реальные сетевые bindings и передачу через relay.
Следующий этап — метрики поставщиков/общей нагрузки, OVERLOAD с подтверждением 500 мс и оптимизатор проб/переноса.

17. Зафиксированный сетевой формат и пределы первого сетевого этапа

Все stream-числа на wire — big-endian, без packed-структур и зависимости от host byte order.
Заголовок raw (52 байта): cmd:1, version:1, command:1, subblock:1,
group_id:8, source_node_id:8, source_epoch:8, sender_epoch:8, receiver_epoch:8, handle:8.
INFO_REQ/RESP согласуют sender/receiver epochs; неизвестная receiver_epoch=0 допустима только при начальном INFO_REQ/reverse.
Подписанный routed payload содержит тот же заголовок; его group_id сверяется с реальной группой router,
sender берётся из end-to-end router metadata. DATA через routed service отвергается.

INFO_RESP/ACCEPT: actual_seq:8, first_available:8, last_available:8, n:1, r:1, block_size:2,
path_count:1, node_ids:8*path_count. В INFO actual_seq — последний доступный, в ACCEPT — фактический старт.
SUBSCRIBE: requested_seq:8; REJECT: reason:1 и для отказа SUBSCRIBE first_available:8,last_available:8.
RENEW: один header handle без payload (каждая кодограмма продлевает один элемент списка).
UNSUBSCRIBE/CLOSED: reason:1. DATA: path_count:1,node_ids:8*path_count,неизменяемый подписанный блок.
GAP: old_cursor:8,next_cursor:8, уведомление включает диапазон [old_cursor,next_cursor).
END: "stream_end":10,group_id:8,source_node_id:8,source_epoch:8,last_seq:8,Ed25519:64 — всего 106 байт.
UINT64_MAX в END означает пустой источник; запрос с конца пустого кольца явно отвергается (нет первого блока).

Предел пути — 32 узла. Сохраняются только реальные узлы пути; исходный {source} не требует отдельной аллокации.
Память пути отделена от 32-МБ бюджета данных и ограничена числом блоков кольца:
до (8 + 8*32) байт на блок плюс поля индекса и overhead allocator; источнику эта память путей не нужна.
64 stream-пира, 256 активных выдач всего и 64 на пира. Control FIFO — до 128 кодограмм,
каждая не больше 338 байт wire; DATA в ней не хранится. На пира один waiter общего send_input_q,
отправка по освобождению до пустого порога; курсоры downstream обслуживаются round-robin.
Эти пределы задают ограниченную память базовой выдачи. A/D/X/Q, возраст нагрузки и 500-мс OVERLOAD
подключаются следующим этапом; автоматический выбор/перекрытие поставщиков пока не запускаются.