Isolate stream bootstrap teardown from shared test events

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);