From d99b0e55253378d0b15b5abf806a5dd39c6fb390 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 17:24:50 +0200 Subject: [PATCH] Reproduce automatic stream stall after early END with incomplete FEC batch --- tests/test_stream_service.c | 34 ++++++++++++++++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index c9a5a296..8d495bc0 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -50,6 +50,7 @@ static unsigned global_data[4]; static unsigned automatic_data[4], automatic_gaps[4], automatic_end[4], automatic_probes[4], automatic_accepts; static uint64_t automatic_next[4]; 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]; err_t __real_uasync_get_load(struct UASYNC *async, struct uasync_load *load); @@ -79,6 +80,9 @@ static void automatic_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { if (automatic_drop_relay && node == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_DATA) { queue_dgram_free(entry); queue_entry_free(entry); return; } + if (automatic_drop_tail && node == 2 && entry->dgram[3] && (command == STREAM_CMD_DATA || command == STREAM_CMD_END)) { + queue_dgram_free(entry); queue_entry_free(entry); return; + } } automatic_handlers[node](conn, entry); } @@ -87,7 +91,7 @@ static void on_automatic(struct UTUN_INSTANCE *inst, const struct stream_event * (void)arg; unsigned node = inst == nodes[1] ? 1 : 2; CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2); if (event->type == STREAM_EVENT_DATA) { - CHECK(event->length == 1180 && !memcmp(event->data, pressure_payload, 1180)); + CHECK(event->length == stream_input_size(&event->descriptor) && !memcmp(event->data, pressure_payload, event->length)); CHECK(event->first_seq == automatic_next[node]++); automatic_data[node]++; } else if (event->type == STREAM_EVENT_GAP) { CHECK(event->first_seq == automatic_next[node] && event->last_seq >= event->first_seq); @@ -722,9 +726,35 @@ static void test_automatic(void) { CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[2], 2); CHECK(automatic_data[2] == 400); stream_close(subscribers[2]); stream_close(publisher); test_clock_run_for(ua, 1000); + /* END первого субпотока не выключает выбор недостающих: для 3+1 нужны ещё данные второй выдачи. */ + parameters.n = 3; + CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); + CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + for (unsigned i = 1; i < 3; i++) { + wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); + CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK); + automatic_next[i] = 0; + } + automatic_drop_tail = 1; + test_clock_run_for(ua, 10000); + size_t input_size = stream_input_size(&origin.descriptor); + memset(pressure_payload, 0xa9, input_size); + CHECK(stream_publish_write(publisher, pressure_payload, input_size, &accepted) == STREAM_OK && accepted == input_size); + CHECK(stream_publish_finish(publisher) == STREAM_OK); + struct stream_local terminal; + deadline = get_time_tb() + 100000; + do { test_clock_poll(ua, 100); CHECK(stream_local_get(nodes[2], &origin.descriptor, &terminal) == STREAM_OK); } + while (!terminal.finished && get_time_tb() < deadline); + CHECK(terminal.finished && terminal.anchored && !terminal.next && automatic_data[2] == 400 && automatic_end[2] == 2); + automatic_drop_tail = 0; + run_until(&automatic_data[2], 401); run_until(&automatic_end[2], 3); + CHECK(automatic_next[2] == 1 && !automatic_gaps[2]); + for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher); + test_clock_run_for(ua, 1000); for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_handlers[i]) == 0); stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only empty source, shared owners, 30s cycle, FEC spread, overlap, stall and CLOSED failover"); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only, shared owners, 30s cycle, FEC spread, overlap, stall, CLOSED and early END drain"); } /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */