diff --git a/doc/stream.txt b/doc/stream.txt index ea51f4ab..53afce3c 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -537,7 +537,7 @@ Stop/destroy выполняются по правилам ядра, вне callb Лог общего прогона: /tmp/stream-service-check.log (107 с). [+] Сетевая выдача — 11.10.2026: - stream_wire.c/h — big-endian формат, raw ETCP_ID_STREAM=0x07, routed ETCP_RT_ID_STREAM=0x39, + stream_wire.c/h — big-endian формат, raw ETCP_ID_STREAM=0x07, routed ETCP_RT_ID_STREAM=0x3C, INFO/ACCEPT с границами окна и путём, 64-битные эпохи/handles, отдельная подпись источника на END. stream_network.c/h — собственный NCD owner пира, общий для его потоков/групп; ожидание его UP, прямые SUBSCRIBE/ACCEPT/REJECT/DATA/GAP/RENEW/UNSUBSCRIBE/CLOSED/END, routed REVERSE_REQ/CLOSED. diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index a923a413..d4dab73e 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -227,7 +227,9 @@ static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id if (!e) return; struct media_super_peer* sp = (struct media_super_peer*)e->data; if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; } - if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; } + if (sp->connection) { conn_mgr_close(sp->connection); sp->connection = NULL; } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: super peer released node=%llx group=%llx", MD_ID, + (unsigned long long)sp->peer_node_id, (unsigned long long)sp->group_id); queue_remove_data(md->super_peers, e); queue_entry_free(e); } @@ -345,18 +347,6 @@ static void md_repl_schedule_retrans(struct media_delivery_ctx* md, struct media md_repl_retrans_cb, "md_repl_retrans"); } -static void md_reconnect_cb(void* arg) { - struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg; - struct media_super_peer* peer = (struct media_super_peer*)md->super_peers; - while (peer) { - if (!peer->connected && peer->connect_timer) { - peer->connect_timer = NULL; - md_super_connect(md, peer->peer_node_id, peer->group_id); - } - peer = peer->connected ? NULL : peer; - } -} - /* ── replication send ── */ static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super_peer* peer) { @@ -687,10 +677,10 @@ static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64 if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: conn_cb for unknown peer 0x%016llx", MD_ID, (unsigned long long)node_id); return; } if (event != CONN_EVENT_UP) { + peer->connected = peer->hello_done = 0; + if (peer->timeout_timer) { uasync_cancel_timeout(md->inst->ua, peer->timeout_timer); peer->timeout_timer = NULL; } DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: connect to supernode 0x%016llx failed rc=%d", MD_ID, (unsigned long long)node_id, event); - peer->connect_timer = uasync_set_timeout(md->inst->ua, - MEDIA_RECONNECT_COOLDOWN_TB, md, md_reconnect_cb, "md_reconnect"); return; } @@ -1106,6 +1096,7 @@ void media_delivery_stream_done(struct UTUN_INSTANCE* inst) { static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id) { struct media_super_peer* peer = md_super_peer_add(md, peer_node_id, group_id); if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)peer_node_id); return; } + if (peer->connection) return; struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id); if (!grp || !grp->conn_mgr) { @@ -1115,7 +1106,9 @@ static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_i DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: connecting to supernode 0x%016llx", MD_ID, (unsigned long long)peer_node_id); - conn_mgr_open(grp->instance, grp->group_id, peer_node_id, md_super_conn_cb, md, NULL); + if (conn_mgr_open(grp->instance, grp->group_id, peer_node_id, md_super_conn_cb, md, &peer->connection) < 0) + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: supernode CM request failed node=%llx group=%llx", MD_ID, + (unsigned long long)peer_node_id, (unsigned long long)group_id); } /* ── supernode start/stop ── */ @@ -1166,13 +1159,8 @@ static void md_super_stop(struct media_delivery_ctx* md) { /* clear super_peers */ if (md->super_peers) { - struct ll_entry* se = md->super_peers->head; - while (se) { - struct media_super_peer* sp = (struct media_super_peer*)se->data; - if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; } - if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; } - se = se->next; - } + while (md->super_peers->head) + md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id); queue_free(md->super_peers); md->super_peers = NULL; } @@ -1460,13 +1448,8 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) { } if (md->super_peers) { - struct ll_entry* se = md->super_peers->head; - while (se) { - struct media_super_peer* sp = (struct media_super_peer*)se->data; - if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; } - if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; } - se = se->next; - } + while (md->super_peers->head) + md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id); queue_free(md->super_peers); md->super_peers = NULL; } diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 344121ea..665c1f62 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -41,7 +41,7 @@ struct media_super_peer { uint8_t connected; // 1 = прямое подключение установлено uint8_t hello_done; // 1 = SUPER_HELLO обмен завершён, можно реплицировать void* timeout_timer; // таймер ретрансмита - void* connect_timer; // таймер переподключения (1 час при ошибке) + struct CONN_MGR_HANDLE* connection; // собственное CM-владение; восстановлением пути управляет CM struct media_delivery_ctx* md; // обратная ссылка }; diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index defd0527..df2a75c7 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1040,7 +1040,7 @@ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_eve /* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */ static void test_supernode_relay(void) { int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE]; - debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_TRACE); + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO); int diagnostic_level = g_debug_config.category_levels[DEBUG_CATEGORY_DEBUG]; debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG); struct TOPO_PEER_REQUEST *connections[3] = {0}; @@ -1238,6 +1238,11 @@ int main(int argc, char **argv) { 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(); + if (u_get_allocated_count() != allocations) { + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR); + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay tracked allocations before=%zu after=%zu", allocations, u_get_allocated_count()); + u_report_unfreed_blocks(); + } CHECK(u_get_allocated_count() == allocations); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] isolated supernode relay checks complete, no tracked allocations remain"); return 0;