Remove closing router channels before notifying services

src/routing_layer/etcp_router.c

@@ -561,12 +561,12 @@ static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) {
     if (!rconn || rconn->closed) return;
     struct UTUN_INSTANCE* inst = rconn->inst;
     rconn->closed = 1;
+    // Callback может закрыть группу или открыть новый канал с тем же ключом.
+    if (inst->router_conns) queue_remove_data(inst->router_conns, &rconn->ll);
     router_send_close_to_service(rconn);
     // Отправляем CLOSE удалённой стороне + снимаем waiter отправки
     if (router_send_conn(rconn)) router_send_ctrl(rconn, ROUTER_FLAG_CLOSE);
     router_conn_free_queues(rconn);
-    if (inst->router_conns)
-        queue_remove_data(inst->router_conns, &rconn->ll);
     DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: closed group=%016llx remote=%016llx svc_id=%u",
                (unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id);
     uasync_call_soon(inst->ua, rconn, router_close_finalize);

tests/test_etcp_router_unit.c

@@ -1407,8 +1407,72 @@ static int test_required_crypto(void) {
 }
 
 // ======================== Main ========================
+static unsigned nested_close_notifications;
+static int nested_close_entered;
+
+/* Уведомление одного канала закрывает остальные каналы той же группы. */
+static void close_group_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
+    uint64_t group;
+    memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, sizeof(group));
+    nested_close_notifications++;
+    queue_dgram_free(entry); queue_entry_free(entry);
+    if (!nested_close_entered) {
+        nested_close_entered = 1;
+        DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "[regression] entering group close from channel-close callback");
+        etcp_router_close_group(conn->instance, group);
+        DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "[regression] returned from nested group close");
+    }
+}
+
+static int test_close_group_from_close_callback(void) {
+    TEST("close callback can close its group without revisiting the closing channel");
+    struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
+    SETUP(); nested_close_notifications = 0; nested_close_entered = 0;
+    debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO);
+    TESTASSERT(!etcp_router_bind(&inst, TEST_SVC_ID, close_group_handler));
+    TESTASSERT(!etcp_router_bind(&inst, TEST_SVC_ID2, close_group_handler));
+    struct ETCP_ROUTER_CONN* first = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID);
+    TESTASSERT(first && etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID2));
+    TESTASSERT(etcp_router_conn_get(&inst, 123, TEST_REMOTE_NODE, TEST_SVC_ID));
+    etcp_router_conn_close(first);
+    TESTASSERT(nested_close_notifications == 2 && inst.router_conns->count == 1);
+    TEARDOWN(); PASS(); return 0;
+}
+
+static struct ETCP_ROUTER_CONN* replacement_channel;
+static int reopen_on_close;
+
+/* Новое соединение получает собственные очереди и эпоху, пока старое ещё завершает callback. */
+static void reopen_channel_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
+    uint64_t group, peer;
+    memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, sizeof(group));
+    memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, sizeof(peer));
+    uint8_t svc = entry->dgram[0];
+    queue_dgram_free(entry); queue_entry_free(entry);
+    if (reopen_on_close) {
+        reopen_on_close = 0;
+        replacement_channel = etcp_router_conn_get(conn->instance, group, peer, svc);
+    }
+}
+
+static int test_reopen_from_close_callback(void) {
+    TEST("close callback can open a fresh channel with the same key");
+    struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
+    SETUP(); replacement_channel = NULL; reopen_on_close = 1;
+    TESTASSERT(!etcp_router_bind(&inst, TEST_SVC_ID, reopen_channel_handler));
+    struct ETCP_ROUTER_CONN* old = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID);
+    TESTASSERT(old); uint64_t epoch = old->reset_id;
+    etcp_router_conn_close(old);
+    TESTASSERT(replacement_channel && replacement_channel != old && !replacement_channel->closed);
+    TESTASSERT(replacement_channel->reset_id != epoch && inst.router_conns->count == 1);
+    TESTASSERT(etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID) == replacement_channel);
+    TEARDOWN(); PASS(); return 0;
+}
+
 int main(void) {
     // Run all tests
+    test_close_group_from_close_callback();
+    test_reopen_from_close_callback();
     test_init_destroy();
     test_bind_unbind();
     test_conn_get_create();