From 3adc4a0be40ec5a832653e95be1204bafb8ca49e Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 00:55:12 +0200 Subject: [PATCH] Isolate stream bootstrap teardown from shared test events --- tests/test_stream_service.c | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 5aaf2e8a..77137296 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -345,6 +345,9 @@ static void bootstrap_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { static void bootstrap_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { if (event->descriptor.source_node_id != nodes[3]->node_id) { on_stream(inst, event, arg); return; } + if (event->type == STREAM_EVENT_CLOSED) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test local reader closed reason=%d", event->reason); return; + } CHECK(event->descriptor.group_id == GROUP1 && event->type == STREAM_EVENT_DATA && event->length == sizeof(expected)); CHECK(!memcmp(event->data, expected, sizeof(expected))); bootstrap.data++; } @@ -424,15 +427,32 @@ static void test_bootstrap(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated REINIT keeps original pending deadline: elapsed_ms=%llu", (unsigned long long)((get_time_tb() - started) / 10)); + stream_close(subscriber); stream_close(publisher); + test_clock_run_for(ua, 1000); /* Доставить отложенные CLOSED до возврата общего обработчика стенда. */ + CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, bootstrap.receive) == 0); + etcp_remove_conn_status_cbk(nodes[1], bootstrap_status, NULL); + stream_set_event_callback(nodes[1], on_stream, NULL); + debug_set_category_level(DEBUG_CATEGORY_ETCP, transport_level); +} + +/* Разрушающий транспорт сценарий выполняется после проверок, которым нужен готовый общий стенд. */ +static void test_pending_down(void) { + struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL; + CHECK(stream_publish_open(nodes[3], GROUP1, NULL, &publisher) == STREAM_OK); + struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + wait_state(1, 3, GROUP1, origin.descriptor.source_epoch, 0); + stream_set_event_callback(nodes[1], bootstrap_event, NULL); + CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &subscriber) == STREAM_OK); + struct STREAM_RECEIVE *probe = NULL; CHECK(stream_probe_open(subscriber, nodes[3]->node_id, 0, &probe) == STREAM_OK); + struct ETCP_CONN *conn = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(conn && conn->links_up); etcp_connection_close(conn); /* Реальный DOWN и DELETE проходят через NCD. */ + struct stream_receive_info info; CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_TRANSPORT); stream_receive_close(probe); stream_close(subscriber); stream_close(publisher); - CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, bootstrap.receive) == 0); - etcp_remove_conn_status_cbk(nodes[1], bootstrap_status, NULL); + test_clock_run_for(ua, 1000); stream_set_event_callback(nodes[1], on_stream, NULL); - debug_set_category_level(DEBUG_CATEGORY_ETCP, transport_level); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] transport DOWN/DELETE still cancels pending request"); } @@ -1091,6 +1111,7 @@ int main(void) { test_pressure(); test_global_pressure(); test_automatic(); + test_pending_down(); /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */ struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);