From eda8dfbc91f777eca5b044236e8e4fc7aef5abbf Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 16:32:53 +0200 Subject: [PATCH] Cover paused sources and duplicate relay controls during startup --- doc/stream.txt | 19 +++++++++++++++---- doc/tasks.md | 8 ++++++++ tests/test_stream_service.c | 26 +++++++++++++++++++++++--- 3 files changed, 46 insertions(+), 7 deletions(-) diff --git a/doc/stream.txt b/doc/stream.txt index f99c499f..81077d57 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -882,8 +882,14 @@ test_stream_service также проверяет общий каталог дв 21. Суперузлы: помощь в раздаче по запросу -Сам старт публикации не запускает приём на суперузлах. Selector зрителя запрашивает помощь после 5 секунд -недостатка действующих субпотоков для k=n-r. Для остановившихся выдач требуется также свидетельство более свежих данных; +Сам старт публикации не запускает приём на суперузлах. Новая подписка сразу запускает INFO-пробы и отдельный +дедлайн 100 мс. Если пригодного ответа ещё нет, запрашивается один суперузел. Первый валидный INFO с разрешённой +выдачей выигрывает: порядок приёма фиксируется локальным счётчиком, независимо от RTT и порядка списка кандидатов. +Проигравшие пробы и неподключённая помощь отменяются. DATA сначала открывается только у победителя; проверенный +первый блок фиксирует общий старт всех остальных субпотоков. До первого восстановления FEC выбирается победитель, +затем действует обычный оптимизатор. Пустой доступный источник не требует помощи, пока он не публикует данные. +Для восстановления работающего потока сохраняются 5 секунд недостатка действующих субпотоков для k=n-r. +Для остановившихся выдач требуется также свидетельство более свежих данных; пауза источника, отсутствие полной FEC-защиты и недобор всех n сами по себе не запускают помощь. Выбирается один онлайн-суперузел данной CHAT-группы с допустимой NCD-попыткой; после неуспеха пробуется следующий. Канальная роль берётся из принятого блока владельца adm_tags.supernode, не из client_type/storage/node_type. @@ -906,11 +912,15 @@ STATUS: state:1,reason:1,remaining_ms:4; для ENDED дополнительно Состояния: STARTING=1, READY=2, BOUND=3, REJECTED=4, ENDED=5, CANCELLED=6. ENDED в ответе помощи разрешён только для пустого источника (last_seq=UINT64_MAX) и проверяется ключом источника. -Bootstrap ограничен 15 секундами с первого запроса, повторы срок не продлевают. READY означает наличие -реальных блоков хотя бы одного субпотока с допустимым путём к запросившему; окончательные индекс/путь/FEC +Bootstrap ограничен 15 секундами с первого запроса, повторы срок не продлевают. READY означает восстановление +первой FEC-пачки и наличие реальных блоков с допустимым путём к запросившему; окончательные индекс/путь/FEC проверяются INFO/ACCEPT/DATA. Переход к READY даёт отдельные 5 секунд для SUBSCRIBE. READY будит selector и снимает старый backoff данного пира. Подтверждённый поставщик временно допускается и выборкой кандидатов, и receive_open, независимо от задержки его Merkle-опции. +Первое подтверждение поколения и переходы состояний отправляются отложенно сразу; одинаковые повторы ограничены 1 с. +Объединённая работа управления выполняется после packet/NCD callbacks. Общий таймер обслуживает сроки/метрики, +новые INFO/ACCEPT/READY и первый DATA/восстановленный блок будят управление без ожидания следующего тика. +Все таймеры/отложенные вызовы/queue waiters отменяются до освобождения владельца. После первой действующей выдачи запрос становится BOUND; дальнейшую потребность подтверждают обычные RENEW/аренда. Если помощь не даёт прогресса потребителя 15 секунд и набор по-прежнему недостаточен, её выдачи закрываются, selector пробует другого суперузла. Запросы/ответы не продлевают аренду raw-выдач. @@ -937,6 +947,7 @@ Relay-only экземпляр не запрашивает новую помощ Отказы по роли, эпохе, лимитам, таймауту и перегрузке диагностируются с причиной. test_stream_service проверяет отсутствие помощи при прямом успехе/паузе источника, автоматический fallback, +дедлайн 100 мс, обе стороны гонки, поздние INFO/REQUEST проигравшего, дубликаты STATUS и общий старт при n=8/r=2, раздачу нескольким зрителям, общий owner с локальным viewer, остановку при уходе последнего, поздний повтор запроса до/после удаления tombstone, CANCEL до REQUEST, истечение READY без SUBSCRIBE, откат неуспешного добавления viewer, исключение relay-only кандидатов и передачу пустого подписанного END. diff --git a/doc/tasks.md b/doc/tasks.md index 21253824..b30bb791 100644 --- a/doc/tasks.md +++ b/doc/tasks.md @@ -4,6 +4,14 @@ ## Текущие задачи +[ ] **Live: ускорить начальное подключение** — 13.10.2026: + Дедлайн помощи 100 мс и первый пригодный INFO выбирают поставщика; DATA использует один общий старт. + Управление просыпается по событиям, подтверждение поколения и новые состояния помощи не ждут 1 с. + READY требует восстановленной FEC-пачки. Регрессии: прямой/relay победитель, поздние ответы, + отмена до дедлайна, пустой источник, дубликаты, общий индекс 8/2 и освобождение ресурсов. + Начальная очередь аудио прогревает Opus и обрезается по первому KEY; prefill 150 мс начинается после него. + Выполняется общая проверка ядра, desktop и APK. Контракт: `doc/stream.txt`, раздел 21. + [+] **Live: сквозное стерео и три компактные кнопки поворота** — 13.10.2026: CAMCORDER/Opus/компрессор/Android AudioTrack/desktop output используют два канала, Opus AUDIO 128kbps. audio_codec=2 обозначает stereo; 960 кадров на канал дают 1920 PCM-отсчётов. diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 3caf0e8d..206bdc92 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -54,8 +54,8 @@ static int automatic_drop_relay; static int automatic_drop_tail; static etcp_recv_fn automatic_handlers[4]; static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; -static etcp_recv_fn relay_raw_handlers[4], relay_route_handler; -static unsigned relay_data[3], relay_ends[3], relay_requests_seen; +static etcp_recv_fn relay_raw_handlers[4], relay_route_handler, relay_status_handler; +static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen; static int relay_drop_source; static uint8_t relay_request_wire[STREAM_WIRE_HEADER]; static struct { @@ -1054,6 +1054,13 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { relay_route_handler(conn, entry); } +/* Счётчик STATUS выявляет ответы на дубликаты, не меняя подписанное управление и его порядок. */ +static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER + && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) relay_statuses_seen++; + relay_status_handler(conn, entry); +} + /* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { (void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3); @@ -1100,15 +1107,22 @@ static void test_supernode_relay(void) { } relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0); + relay_status_handler = nodes[2]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; + CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_test_status) == 0); struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0}; CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + test_clock_run_for(ua, 10000); + struct stream_local local; + CHECK(!relay_requests_seen && !relay_data[2]); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000); CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK); CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); run_until(&relay_data[2], 1); test_clock_run_for(ua, 60000); - struct stream_local local; CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen); stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 10000); @@ -1149,6 +1163,11 @@ static void test_supernode_relay(void) { CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes); relay_release_info(); test_clock_run_for(ua, 1000); CHECK(!relay_race.source_subscribes && relay_data[2] == 3); + struct stream_message duplicate; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &duplicate) == STREAM_OK); + CHECK(duplicate.receiver_epoch); + unsigned statuses_before = relay_statuses_seen; + for (unsigned i = 0; i < 3; i++) { relay_test_control(&duplicate); test_clock_run_for(ua, 300); } + CHECK(relay_statuses_seen == statuses_before); relay_race.check_starts = 0; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap helper wins: elapsed_ms=%llu, n=8 r=2, common_start=%llu, stale direct INFO ignored", (unsigned long long)connected_ms, (unsigned long long)relay_race.start); @@ -1236,6 +1255,7 @@ static void test_supernode_relay(void) { stream_close(viewers[2]); stream_close(publisher); relay_drop_source = 0; relay_role(1, 0, 101); CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0); + CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_status_handler) == 0); for (unsigned i = 0; i < 4; i++) { CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0); if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);