Move direct transport tests to NCD ownership

tests/test_etcp_connect.c

@@ -36,6 +36,7 @@ static char tdir[] = "/tmp/utun_ec_XXXXXX";
 static char ca[256], cb[256];
 static int pa = 0, pb = 0;
 static struct NODE_CONN_DIRECT *chat_a = NULL, *chat_b = NULL;
+static struct NODE_CONN_DIRECT *attempts[2];
 
 static int wf(const char* p, const char* f, ...) {
     va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1;
@@ -67,7 +68,9 @@ static void test11(void* arg);
 /* ----- callback state per test ----- */
 static volatile int cb_type = 0, cb_conn_ok = 0, cb_count = 0;
 
-static void connect_ccb(void* a, struct ETCP_CONN* conn, int type) { cb_type = type; cb_conn_ok = (conn != NULL); cb_count++; }
+static void connect_ccb(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
+    (void)arg; cb_type = event; cb_conn_ok = node_conn_direct_get_conn(h) != NULL; cb_count++;
+}
 
 /* ----- helper: create minimal TOPO_GROUP_NODE with 1 v4 addr ----- */
 static struct TOPO_GROUP_NODE* mknode(struct UTUN_INSTANCE* inst, uint64_t nid, const uint8_t pubkey[32],
@@ -87,24 +90,25 @@ static struct TOPO_GROUP_NODE* mknode(struct UTUN_INSTANCE* inst, uint64_t nid,
     return nq;
 }
 
-/* ======================== Test 1: already connected → immediate EARLY+LATE ======================== */
+/* Готовый транспорт: UP отложен, handle доступен до callback. */
 static void test1(void* arg) {
     (void)arg; if (result) return;
     if (lks(g_a) < 1 || lks(g_b) < 1) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test1, "t1"); return; }
     struct TOPO_GROUP_NODE* nb = topo_node_find_by_id(topo_groups_get_default(g_a->topo_groups), nid_b);
     if (!nb) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test1, "t1b"); return; }
-    fprintf(stderr, "Test 1: already connected → EARLY|LATE\n"); fflush(stderr);
+    fprintf(stderr, "Test 1: existing transport, deferred UP\n"); fflush(stderr);
     cb_type = 0; cb_conn_ok = 0; cb_count = 0;
-    int r = etcp_connect(g_a, nb, connect_ccb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE);
-    if (r != 0) { fail("etcp_connect returned error"); return; }
+    int r = node_conn_direct_open(g_a, nb->node_id, connect_ccb, NULL, &attempts[0], NULL);
+    if (r != NCD_REUSED || !attempts[0] || cb_count) { fail("NCD must return handle before deferred UP"); return; }
     uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test2, "t2");
 }
 static void test2(void* arg) {
     (void)arg; if (result) return;
     if (!cb_count) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test2, "t2"); return; }
-    if (cb_count != 2) { fail("test1: expected 2 callbacks"); return; }
+    if (cb_count != 1 || cb_type != NCD_EVENT_UP) { fail("test1: expected one UP"); return; }
     if (!cb_conn_ok) { fail("test1: conn=NULL in callback"); return; }
     fprintf(stderr, "  OK: %d callbacks, conn=OK\n", cb_count); fflush(stderr);
+    node_conn_direct_close(attempts[0]); attempts[0] = NULL;
     /* advance to test3 */
     uasync_call_soon(ua, NULL, test3);
 }
@@ -120,24 +124,24 @@ static void test3(void* arg) {
     queue_data_put_with_index(topo_groups_get_default(g_a->topo_groups)->nodes, &fn->ll);
     g_a->etcp_connect_timeout_tb = 2000;
     cb_type = -1; cb_conn_ok = -1; cb_count = 0;
-    int r = etcp_connect(g_a, fn, connect_ccb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE);
-    if (r != 0) { fail("etcp_connect(unreachable) returned error"); return; }
+    int r = node_conn_direct_open(g_a, fn->node_id, connect_ccb, NULL, &attempts[0], NULL);
+    if (r < 0) { fail("NCD unreachable attempt rejected"); return; }
     uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test4, "t4");
 }
 static void test4(void* arg) {
     (void)arg; if (result) return;
     if (!cb_count) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test4, "t4"); return; }
     if (cb_count != 1) { fail("test3: expected 1 callback"); return; }
-    if (cb_conn_ok != 0) { fail("test3: expected conn=NULL"); return; }
-    if (cb_type != 0) { fail("test3: expected type=0 (error)"); return; }
-    fprintf(stderr, "  OK: error callback delivered (NULL, type=0)\n"); fflush(stderr);
+    if (!cb_conn_ok || cb_type != NCD_EVENT_TIMEOUT) { fail("test3: expected TIMEOUT with retained transport"); return; }
+    node_conn_direct_close(attempts[0]); attempts[0] = NULL;
+    fprintf(stderr, "  OK: TIMEOUT delivered, owner released\n"); fflush(stderr);
     uasync_call_soon(ua, NULL, test5);
 }
 
 /* ======================== Test 5: double connect to unreachable → both callbacks ======================== */
 static int t5_cb_count = 0;
-static void t5_ccb(void* a, struct ETCP_CONN* conn, int type) {
-    (void)a; (void)conn; if (type == 0) t5_cb_count++;
+static void t5_ccb(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
+    (void)h; (void)arg; if (event == NCD_EVENT_TIMEOUT) t5_cb_count++;
 }
 
 static void test5(void* arg) {
@@ -150,9 +154,11 @@ static void test5(void* arg) {
     queue_data_put_with_index(topo_groups_get_default(g_a->topo_groups)->nodes, &fn->ll);
     g_a->etcp_connect_timeout_tb = 2000;
     t5_cb_count = 0;
-    int r1 = etcp_connect(g_a, fn, t5_ccb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE);
-    int r2 = etcp_connect(g_a, fn, t5_ccb, NULL, ETCP_CONNECT_EARLY);
-    if (r1 != 0 || r2 != 0) { fail("etcp_connect double call returned error"); return; }
+    int r1 = node_conn_direct_open(g_a, fn->node_id, t5_ccb, NULL, &attempts[0], NULL);
+    int r2 = node_conn_direct_open(g_a, fn->node_id, t5_ccb, NULL, &attempts[1], NULL);
+    if (r1 < 0 || r2 != NCD_REUSED || node_conn_direct_get_conn(attempts[0]) != node_conn_direct_get_conn(attempts[1])) {
+        fail("NCD attempts must share transport"); return;
+    }
     uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test6, "t6");
 }
 static void test6(void* arg) {
@@ -160,46 +166,52 @@ static void test6(void* arg) {
     if (t5_cb_count < 2) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test6, "t6"); return; }
     if (t5_cb_count != 2) { fail("test5: expected 2 error callbacks"); return; }
     fprintf(stderr, "  OK: both callbacks fired\n"); fflush(stderr);
+    for (int i = 0; i < 2; i++) { node_conn_direct_close(attempts[i]); attempts[i] = NULL; }
     uasync_call_soon(ua, NULL, test7);
 }
 
-/* ======================== Test 7: EARLY-only flag ======================== */
+/* Закрытие handle отменяет его отложенный UP, второй владелец получает один UP. */
 static void test7(void* arg) {
     (void)arg; if (result) return;
     struct TOPO_GROUP_NODE* nb = topo_node_find_by_id(topo_groups_get_default(g_a->topo_groups), nid_b);
     if (!nb || lks(g_a) < 1) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test7, "t7"); return; }
-    fprintf(stderr, "Test 7: EARLY-only flag\n"); fflush(stderr);
+    fprintf(stderr, "Test 7: cancel deferred UP independently\n"); fflush(stderr);
     cb_type = 0; cb_conn_ok = 0; cb_count = 0;
-    int r = etcp_connect(g_a, nb, connect_ccb, NULL, ETCP_CONNECT_EARLY);
-    if (r != 0) { fail("etcp_connect(EARLY) returned error"); return; }
+    int r = node_conn_direct_open(g_a, nb->node_id, connect_ccb, NULL, &attempts[0], NULL);
+    if (r != NCD_REUSED) { fail("cannot open cancelled owner"); return; }
+    node_conn_direct_close(attempts[0]); attempts[0] = NULL;
+    r = node_conn_direct_open(g_a, nb->node_id, connect_ccb, NULL, &attempts[0], NULL);
+    if (r != NCD_REUSED) { fail("cannot reopen owner"); return; }
     uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test8, "t8");
 }
 static void test8(void* arg) {
     (void)arg; if (result) return;
     if (!cb_count) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test8, "t8"); return; }
     if (cb_count != 1) { fail("test7: expected exactly 1 callback"); return; }
-    if (cb_type != ETCP_CONNECT_EARLY) { fail("test7: expected EARLY type"); return; }
-    fprintf(stderr, "  OK: only EARLY fired\n"); fflush(stderr);
+    if (cb_type != NCD_EVENT_UP) { fail("test7: expected UP"); return; }
+    node_conn_direct_close(attempts[0]); attempts[0] = NULL;
+    fprintf(stderr, "  OK: cancelled callback suppressed\n"); fflush(stderr);
     uasync_call_soon(ua, NULL, test9);
 }
 
-/* ======================== Test 9: LATE-only flag ======================== */
+/* Новый владелец готового транспорта не создаёт второе соединение. */
 static void test9(void* arg) {
     (void)arg; if (result) return;
     struct TOPO_GROUP_NODE* nb = topo_node_find_by_id(topo_groups_get_default(g_a->topo_groups), nid_b);
     if (!nb || lks(g_a) < 1) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test9, "t9"); return; }
-    fprintf(stderr, "Test 9: LATE-only flag\n"); fflush(stderr);
+    fprintf(stderr, "Test 9: reused transport owner\n"); fflush(stderr);
     cb_type = 0; cb_conn_ok = 0; cb_count = 0;
-    int r = etcp_connect(g_a, nb, connect_ccb, NULL, ETCP_CONNECT_LATE);
-    if (r != 0) { fail("etcp_connect(LATE) returned error"); return; }
+    int r = node_conn_direct_open(g_a, nb->node_id, connect_ccb, NULL, &attempts[0], NULL);
+    if (r != NCD_REUSED) { fail("cannot reuse transport"); return; }
     uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test10, "t10");
 }
 static void test10(void* arg) {
     (void)arg; if (result) return;
     if (!cb_count) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test10, "t10"); return; }
     if (cb_count != 1) { fail("test9: expected exactly 1 callback"); return; }
-    if (cb_type != ETCP_CONNECT_LATE) { fail("test9: expected LATE type"); return; }
-    fprintf(stderr, "  OK: only LATE fired\n"); fflush(stderr);
+    if (cb_type != NCD_EVENT_UP) { fail("test9: expected UP"); return; }
+    node_conn_direct_close(attempts[0]); attempts[0] = NULL;
+    fprintf(stderr, "  OK: owner reused transport\n"); fflush(stderr);
     if (node_conn_direct_open(g_a, nid_b, NULL, NULL, &chat_a, NULL) != NCD_REUSED ||
         node_conn_direct_open(g_b, nid_a, NULL, NULL, &chat_b, NULL) != NCD_REUSED) {
         fail("cannot retain shared connection for chat"); return;
@@ -290,6 +302,7 @@ int main(void) {
       while (!result && (int)(get_time_tb() - start) < TIMEOUT_TB + 50000) { uasync_poll(ua, POLL_MS); } }
     fprintf(stderr, "final result=%d\n", result); fflush(stderr);
 done:
+    for (int i = 0; i < 2; i++) if (attempts[i]) node_conn_direct_close(attempts[i]);
     if (ttimer) uasync_cancel_timeout(ua, ttimer);
     if (chat_a) node_conn_direct_close(chat_a);
     if (chat_b) node_conn_direct_close(chat_b);

tests/test_etcp_stcp.c

@@ -81,10 +81,12 @@ static void connection_event(struct ETCP_CONN* conn, int event, void* arg) {
     if (event == ETCP_CBK_EVENT_DELETE) endpoint->deleted++;
 }
 
-static void connect_done(void* arg, struct ETCP_CONN* conn, int type) {
+static void connect_done(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
     struct endpoint* endpoint = arg;
-    if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Connection failed: local=%016llx", (unsigned long long)endpoint->id); failed = 1; }
-    else if (type & ETCP_CONNECT_EARLY) endpoint->conn = conn;
+    if (event == NCD_EVENT_TIMEOUT || event == NCD_EVENT_CLOSED) {
+        DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Connection failed: local=%016llx event=%d", (unsigned long long)endpoint->id, event);
+        failed = 1;
+    } else if (event == NCD_EVENT_UP) endpoint->conn = node_conn_direct_get_conn(h);
 }
 
 static void start_endpoint(int i, int reality) {
@@ -161,7 +163,9 @@ static void check_unique(int i) {
 static void hold_connection(int i) {
     struct endpoint* endpoint = &endpoints[i];
     struct TOPO_NODE* peer = topo_node_registry_find(endpoint->inst->topo_groups, endpoints[1-i].id); CHECK(peer);
-    CHECK(node_conn_direct_open_node(endpoint->inst, peer->node_id, NULL, NULL, &endpoint->owner, peer, NULL) == NCD_REUSED);
+    if (!endpoint->owner)
+        CHECK(node_conn_direct_open_node(endpoint->inst, peer->node_id, NULL, NULL, &endpoint->owner, peer, NULL) == NCD_REUSED);
+    node_conn_direct_set_callback(endpoint->owner, NULL, NULL);
     CHECK(node_conn_direct_get_conn(endpoint->owner) == endpoint->conn);
     etcp_conn_add_cbk(endpoint->conn, connection_event, endpoint, ETCP_CBK_EVENT_REINIT | ETCP_CBK_EVENT_DELETE);
 }
@@ -195,8 +199,8 @@ static void run_case(int reality, int simultaneous, int restarted) {
     start_endpoint(0, reality); start_endpoint(1, reality);
     struct TOPO_NODE* peers[2] = {import_peer(0), import_peer(1)};
     for (int i = 0; i <= simultaneous; i++) {
-        struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id;
-        CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0);
+        CHECK(node_conn_direct_open_node(endpoints[i].inst, peers[i]->node_id, connect_done, &endpoints[i],
+            &endpoints[i].owner, peers[i], NULL) >= 0);
     }
     wait_pair();
     for (int i = 0; i < 2; i++) { check_unique(i); hold_connection(i); }
@@ -235,8 +239,8 @@ static void run_case(int reality, int simultaneous, int restarted) {
         stop_endpoint(0); stop_endpoint(1); start_endpoint(0, reality); start_endpoint(1, reality);
         peers[0] = import_peer(0); peers[1] = import_peer(1);
         for (int i = 0; i < 2; i++) {
-            struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id;
-            CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY) == 0);
+            CHECK(node_conn_direct_open_node(endpoints[i].inst, peers[i]->node_id, connect_done, &endpoints[i],
+                &endpoints[i].owner, peers[i], NULL) >= 0);
         }
         phase++; wait_pair();
         for (int i = 0; i < 2; i++) { CHECK(endpoints[i].conn->reset_id != old_epochs[i]); check_unique(i); }

tests/test_stcp_traffic.c

@@ -3,6 +3,7 @@
 #include "etcp_api.h"
 #include "etcp.h"
 #include "etcp_connections.h"
+#include "node_conn_direct.h"
 #include "secure_channel.h"
 #include "topo_group.h"
 #include "topo_node.h"
@@ -37,6 +38,7 @@ struct rx_ctx {
 static struct rx_ctx srv_rx, cli_rx;
 static struct UTUN_INSTANCE *srv_inst, *cli_inst;
 static struct ETCP_CONN *srv_conn, *cli_conn;
+static struct NODE_CONN_DIRECT *cli_owner;
 static int srv_send_ready, cli_send_ready;
 
 static void srv_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) {
@@ -67,12 +69,14 @@ static void cli_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) {
     if (cli_rx.len >= cli_rx.expected_total) cli_rx.done = 1;
 }
 
-static void connect_cb(void *arg, struct ETCP_CONN *conn, int type) {
+static void connect_cb(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
     (void)arg;
-    if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "etcp_connect failed"); test_failed = 1; return; }
-    if (type & ETCP_CONNECT_EARLY) {
-        DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "etcp_connect: EARLY ready");
-        cli_conn = conn;
+    if (event == NCD_EVENT_TIMEOUT || event == NCD_EVENT_CLOSED) {
+        DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "NCD failed event=%d", event); test_failed = 1; return;
+    }
+    if (event == NCD_EVENT_UP) {
+        DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "NCD transport UP");
+        cli_conn = node_conn_direct_get_conn(h);
         cli_send_ready = 1;
     }
 }
@@ -166,7 +170,7 @@ int main(void) {
       TASSERT(topo_node_deserialize(cli_group, wire, (size_t)len, &cli_ni, NULL, NULL, NULL, NULL) == 0);
       TASSERT(topo_node_registry_store(cli_inst->topo_groups, cli_ni)); }
     cli_inst->etcp_connect_timeout_tb = 100000;
-    TASSERT(etcp_connect(cli_inst, srv_node, connect_cb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0);
+    TASSERT(node_conn_direct_open(cli_inst, srv_node->node_id, connect_cb, NULL, &cli_owner, NULL) >= 0);
 
     int ticks = 0;
     while (!cli_send_ready && ticks < 5000) { uasync_poll(ua, 10); ticks++; }
@@ -219,6 +223,7 @@ int main(void) {
     if (srv_rx.buf) u_free(srv_rx.buf);
     if (cli_rx.buf) u_free(cli_rx.buf);
 
+    node_conn_direct_close(cli_owner); cli_owner = NULL;
     srv_inst->running = 0; cli_inst->running = 0;
     utun_instance_destroy(srv_inst); srv_inst = NULL;
     utun_instance_destroy(cli_inst); cli_inst = NULL;