From d0ddc9ae8c50190d03c1738f76dbe3bdc7fa34b9 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 16:02:24 +0200 Subject: [PATCH] Fix partial stream bootstrap and release drained upstream flows --- src/stream/stream_optimizer.c | 34 ++++++++---- tests/test_stream_service.c | 100 +++++++++++++++++++++++++++++++++- 2 files changed, 120 insertions(+), 14 deletions(-) diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 687a9461..8e1f4c8f 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -259,7 +259,6 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig for (struct stream_supplier *s = o->suppliers; s; s = s->next) { if (!s->online || !s->compatible || s->node == exclude || !s->offers || now < s->retry_ms || (s->probe && s->probe_sub == sub)) continue; - if (!o->startup_complete && o->startup_peer && s->node != o->startup_peer) continue; struct stream_offer *offer = &s->offers[sub]; struct stream_receive_info *info = &offer->info; if (now < offer->retry_ms || !info->info_ms || now - info->info_ms > STREAM_OFFER_TTL_MS || !info->accepting || !info->free_slots || (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)) continue; @@ -337,12 +336,23 @@ int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_desc o->info.automatic = 1; *out = o; return STREAM_OK; } +/* Освобождает основные и заменяющие выдачи; повторный вызов безопасен, статистика истории сохраняется. */ +static unsigned selections_close(struct stream_optimizer *o) { + unsigned closed = 0; + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct stream_selection *pick = &o->selected[sub]; + closed += (pick->current != NULL) + (pick->replacement != NULL); + stream_receive_close(pick->current); stream_receive_close(pick->replacement); + memset(pick, 0, sizeof(*pick)); + } + o->info.subflows = o->info.overlaps = o->info.providers = o->info.protected_by_fec = 0; + return closed; +} + void stream_optimizer_destroy(struct stream_optimizer *o) { if (!o) return; if (o->startup_timer) uasync_cancel_timeout(o->inst->ua, o->startup_timer); - for (unsigned sub = 0; sub < o->descriptor.n; sub++) { - stream_receive_close(o->selected[sub].current); stream_receive_close(o->selected[sub].replacement); - } + selections_close(o); while (o->suppliers) { struct stream_supplier *s = o->suppliers; o->suppliers = s->next; stream_receive_close(s->probe); u_free(s->offers); u_free(s); @@ -396,9 +406,13 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */ if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) { stream_relay_demand(o->inst, &o->descriptor, 0, now); + unsigned closed = selections_close(o); for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; } - if (o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer END drained source=%llx next=%llu last=%llu", - (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last); + if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; } + if (closed || o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, + "stream optimizer END drained source=%llx next=%llu last=%llu released=%u probes=%u", + (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, + (unsigned long long)local.last, closed, o->info.probes_active); o->draining = 0; o->info.probes_active = 0; return; } if (!o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer draining END source=%llx next=%llu last=%llu", @@ -450,10 +464,7 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS && offer->last_available > local.next) newer = 1; } if (newer) { - for (unsigned sub = 0; sub < o->descriptor.n; sub++) { - struct stream_selection *pick = &o->selected[sub]; stream_receive_close(pick->current); stream_receive_close(pick->replacement); - memset(pick, 0, sizeof(*pick)); - } + selections_close(o); if (stream_local_restart(o->inst, &o->descriptor) == STREAM_OK) { o->info.restarts++; o->progress_ms = now; busy = 0; stream_local_get(o->inst, &o->descriptor, &local); } @@ -467,8 +478,7 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, if (s && stream_receive_open(h, s->node, (uint8_t)sub, UINT64_MAX, &o->selected[sub].current) == STREAM_OK) { o->startup_peer = s->node; o->selected[sub].opened = 1; o->selected[sub].opened_ms = now; busy++; if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; } - for (struct stream_supplier *other = o->suppliers; other; other = other->next) - if (other != s && other->probe) { stream_receive_close(other->probe); other->probe = NULL; } + /* Остальные INFO нужны для недостающих частей: победитель определяет только общий DATA старт. */ DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap winner group=%llx source=%llx epoch=%llx peer=%llx sub=%u" " order=%llu elapsed_ms=%llu relay=%d", (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 1fafb391..59945798 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1444,6 +1444,100 @@ static void test_delayed_media_reader(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] delayed media readers: 6s pause, live progress without restart, exact remote/local FIN drain"); } +static struct { + etcp_recv_fn viewer_receive, source_receive; + struct ll_entry *held_info; + struct ETCP_CONN *held_conn; + unsigned data, ends, renews; + int hold_source; +} lifecycle; + +/* Первый INFO источника задержан до DATA неполного ретранслятора; остальные пакеты не теряются. */ +static void lifecycle_view_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (lifecycle.hold_source && conn->peer_node_id == nodes[0]->node_id + && entry->len >= STREAM_WIRE_HEADER && entry->dgram[2] == STREAM_CMD_INFO_RESP) { + CHECK(!lifecycle.held_info); + lifecycle.held_info = entry; lifecycle.held_conn = conn; return; + } + lifecycle.viewer_receive(conn, entry); +} + +static void lifecycle_source_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (conn->peer_node_id == nodes[2]->node_id && entry->len >= STREAM_WIRE_HEADER + && entry->dgram[2] == STREAM_CMD_RENEW) lifecycle.renews++; + lifecycle.source_receive(conn, entry); +} + +static void lifecycle_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)inst; (void)arg; + if (event->type == STREAM_EVENT_DATA) { + CHECK(event->first_seq == 0 && event->length == sizeof(expected)); + CHECK(!memcmp(event->data, expected, sizeof(expected))); lifecycle.data++; + } else if (event->type == STREAM_EVENT_END) lifecycle.ends++; + else CHECK(event->type == STREAM_EVENT_CLOSED); +} + +/* Первый поставщик имеет лишь 3 из 6 нужных частей; пауза источника не оправдывает зависший bootstrap. */ +static void test_partial_bootstrap_and_end(void) { + struct STREAM_HANDLE *publisher = NULL, *relay = NULL, *viewer = NULL; + CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); + struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + wait_state(1, 0, GROUP1, origin.descriptor.source_epoch, 0); + wait_state(2, 0, GROUP1, origin.descriptor.source_epoch, 0); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &relay) == STREAM_OK); + wait_state(2, 1, GROUP1, origin.descriptor.source_epoch, 1); + memset(expected, 0x37, sizeof(expected)); + CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK); + struct stream_local source, received; + CHECK(stream_local_get(nodes[0], &origin.descriptor, &source) == STREAM_OK); + for (unsigned sub = 0; sub < 3; sub++) { + const uint8_t *part; size_t length; uint64_t actual; + CHECK(stream_buffer_read(source.buffer, 0, sub, &part, &length, &actual) == STREAM_OK); + CHECK(stream_ingest_block(nodes[1], part, length) == STREAM_OK); + } + lifecycle.viewer_receive = nodes[2]->api_bindings.callbacks[ETCP_ID_STREAM]; + lifecycle.source_receive = nodes[0]->api_bindings.callbacks[ETCP_ID_STREAM]; + CHECK(etcp_bind(nodes[2], ETCP_ID_STREAM, lifecycle_view_raw) == 0); + CHECK(etcp_bind(nodes[0], ETCP_ID_STREAM, lifecycle_source_raw) == 0); + stream_set_event_callback(nodes[2], lifecycle_event, NULL); lifecycle.hold_source = 1; + CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &viewer) == STREAM_OK); + uint64_t deadline = get_time_tb() + 40000; + do { + test_clock_poll(ua, 100); + CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK); + } while (!received.anchored && get_time_tb() < deadline); + CHECK(received.anchored && !received.decoded && lifecycle.held_info); + lifecycle.hold_source = 0; + struct ll_entry *held = lifecycle.held_info; lifecycle.held_info = NULL; + lifecycle.viewer_receive(lifecycle.held_conn, held); + run_until(&lifecycle.data, 1); + CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK); + CHECK(received.start == 0 && received.next == 1 && received.decoded); + struct stream_selection_info selected; CHECK(stream_selection_get_info(viewer, &selected) == STREAM_OK); + CHECK(!selected.restarts && selected.providers == 2); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] partial bootstrap: complementary source parts decoded original batch without restart"); + /* END не зависит от существования relay; оставляем только прямую выдачу, чтобы измерять её RENEW. */ + stream_close(viewer); stream_close(relay); viewer = NULL; + test_clock_run_for(ua, 10000); + CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &viewer) == STREAM_OK); + run_until(&lifecycle.data, 2); test_clock_run_for(ua, 20000); + CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&lifecycle.ends, 1); + test_clock_run_for(ua, 2000); lifecycle.renews = 0; + test_clock_run_for(ua, 100000); + CHECK(stream_selection_get_info(viewer, &selected) == STREAM_OK); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "finished reader retained: renews=%u flows=%u overlaps=%u probes=%u providers=%u", + lifecycle.renews, selected.subflows, selected.overlaps, selected.probes_active, selected.providers); + CHECK(!lifecycle.renews && !selected.subflows && !selected.overlaps && !selected.probes_active && !selected.providers); + CHECK(stream_handle_active(viewer)); + CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK && received.next == 1 && received.finished); + struct ETCP_CONN *shared = instance_find_conn(nodes[2], nodes[0]->node_id); CHECK(shared && shared->links_up); + stream_close(viewer); stream_close(publisher); test_clock_run_for(ua, 1000); + CHECK(etcp_bind(nodes[2], ETCP_ID_STREAM, lifecycle.viewer_receive) == 0); + CHECK(etcp_bind(nodes[0], ETCP_ID_STREAM, lifecycle.source_receive) == 0); + stream_set_event_callback(nodes[2], NULL, NULL); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] completed receive releases upstream while retaining reader and shared transport"); +} + int main(int argc, char **argv) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); @@ -1508,8 +1602,9 @@ int main(int argc, char **argv) { CHECK(member_sync_start(nodes[0], nodes[i]->node_id, "710001", NULL, NULL) == 0); CHECK(member_sync_start(nodes[i], nodes[0]->node_id, "710001", NULL, NULL) == 0); } - if (argc == 2 && !strcmp(argv[1], "--relay-only")) { - test_supernode_relay(); + if (argc == 2 && (!strcmp(argv[1], "--relay-only") || !strcmp(argv[1], "--lifecycle-only"))) { + if (!strcmp(argv[1], "--relay-only")) test_supernode_relay(); + else test_partial_bootstrap_and_end(); for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); for (unsigned i = 0; i < 4; i++) { stream_set_event_callback(nodes[i], NULL, NULL); utun_instance_destroy(nodes[i]); } uasync_destroy(ua, 1); test_clock_finish(); @@ -1644,6 +1739,7 @@ int main(int argc, char **argv) { CHECK(data_events[0] == 1); /* Pending 100ms batch не выполнилась после stop. */ DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] CHAT stop keeps UTUN/core; source timer cancelled; stale signed options cleared on restart"); for (unsigned i = 0; i < 4; i++) stream_set_event_callback(nodes[i], NULL, NULL); + test_partial_bootstrap_and_end(); test_delayed_media_reader(); for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); for (unsigned i = 0; i < 4; i++) utun_instance_destroy(nodes[i]);