diff --git a/src/transport_layer/node_conn_direct.c b/src/transport_layer/node_conn_direct.c index db46f0b6..ddb9e0bb 100644 --- a/src/transport_layer/node_conn_direct.c +++ b/src/transport_layer/node_conn_direct.c @@ -148,42 +148,39 @@ static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t no return NULL; } -/* ═══════════ создание линков (round‑robin, все сокеты кроме PRIVATE) ═══════════ */ -/* - * Создаёт по одному ETCP_LINK на каждый адрес пира (IPv4 и IPv6). - * - * Все адреса пира распределяются round-robin по локальным сокетам. - * Если specific_sock задан — использует только его. PRIVATE-сокеты пропускаются. - * - * Особые случаи: - * - Нулевые адреса и нулевые порты пропускаются - * - Не-UDP протоколы пропускаются - * - IPv6 link-local: автоматически выставляется scope_id по netif_index сокета - * - Stale-линки: если на том же addr:port висит старый conn с другим pubkey - * и старый conn ещё не поднялся — stale-линк вытесняется (узел пересоздался - * с новым ключом, старый conn больше не нужен) - */ +/* Все разрешённые пары адрес/сокет: UDP, TCP, IPv4/IPv6 и REALITY. + * Явный specific_sock в open обходит локальную политику, но не проверку семейства/протокола/цели. */ /* Создаёт один UDP-линк на use_sock → sa, с вытеснением stale-линка и дедупом. */ -#define NCD_MAX_SOCKET_VIEWS 32 +#define NCD_MAX_SOCKET_VIEWS 256 /* sock_id уникален среди живых сокетов и имеет тип uint8_t. */ static int ncd_add_udp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* use_sock, struct sockaddr_storage* sa) { struct ETCP_LINK* stale = etcp_link_find_by_addr(use_sock, sa, 0); - if (stale && stale->etcp == conn) return 0; + if (stale && stale->etcp == conn) { + if (!stale->is_server && !stale->link_status && !stale->init_timer) { + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] resume idle UDP probe node=%016llx link=%u", + (unsigned long long)conn->peer_node_id, stale->local_link_id); + etcp_link_enter_init(stale); + } + return conn->close_requested ? -1 : 0; + } if (stale && stale->etcp != conn && memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE)) { if (!stale->etcp->links_up || !stale->etcp->initialized) { struct ETCP_CONN* old = stale->etcp; - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[ncd] evict stale link node=0x%016llx(old pk=%016llx) for new node=0x%016llx(pk=%016llx)", - (unsigned long long)old->peer_node_id, *(const uint64_t*)old->crypto_ctx.peer_public_key, - (unsigned long long)conn->peer_node_id, *(const uint64_t*)conn->crypto_ctx.peer_public_key); + if (etcp_conn_ref_take(old) < 0) return -1; + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] evict stale endpoint=%s old_node=%016llx new_node=%016llx", + sockaddr_storage_to_str(sa).str, (unsigned long long)old->peer_node_id, + (unsigned long long)conn->peer_node_id); etcp_link_close(stale); - etcp_cbk_fire(old, ETCP_CBK_EVENT_NODE_CHANGED); + if (!old->close_requested) etcp_cbk_fire(old, ETCP_CBK_EVENT_NODE_CHANGED); + etcp_conn_ref_free(old); + if (conn->close_requested) return -1; stale = NULL; } else { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[ncd] addr conflict: port already used by live conn 0x%016llx for node 0x%016llx", (unsigned long long)stale->etcp->peer_node_id, (unsigned long long)conn->peer_node_id); - return 0; + return -1; } } if (stale && stale->etcp != conn @@ -195,9 +192,12 @@ static int ncd_add_udp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* use_sock (void*)conn, conn->log_name, conn->state, (void*)stale->etcp, stale->etcp->log_name, stale->etcp->state, stale->etcp->initialized, stale->etcp->links_up); - return 0; + return -1; } - return etcp_link_new(conn, use_sock, sa, 0) ? 1 : 0; + if (etcp_link_new(conn, use_sock, sa, 0)) return 1; + DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] UDP link creation failed node=%016llx address=%s", + (unsigned long long)conn->peer_node_id, sockaddr_storage_to_str(sa).str); + return -1; } static int ncd_same_address(const struct sockaddr_storage* a, const struct sockaddr_storage* b) { @@ -220,6 +220,7 @@ static int ncd_add_tcp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* sock, struct sockaddr_storage* addr, const struct TOPO_REALITY_SOCK* reality) { for (struct ETCP_LINK* link = conn->links; link; link = link->next) { if (!link->is_tcp || link->is_server || link->conn != sock || !ncd_same_address(&link->remote_addr, addr)) continue; + link->active_users++; struct reality_client_config before; memcpy(&before, &link->reality, sizeof(before)); int had_reality = link->reality_set; if (reality) topo_node_apply_reality(link, reality); @@ -229,23 +230,28 @@ static int ncd_add_tcp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* sock, (unsigned long long)conn->peer_node_id, link->local_link_id, link->link_status); if (!link->link_status) etcp_tcp_link_start_reconnect(link); } - if (!link->link_status && !link->tcp_link && !link->tcp_reconnect_timer) { + if (!link->closing && !conn->close_requested && !link->link_status && !link->tcp_link && !link->tcp_reconnect_timer) { DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] resume idle TCP probe node=%016llx link=%u", (unsigned long long)conn->peer_node_id, link->local_link_id); etcp_tcp_link_start_reconnect(link); } - return 0; + int cancelled = link->closing || conn->close_requested; + etcp_link_release(link); + return cancelled ? -1 : 0; } struct ETCP_LINK* link = etcp_link_new(conn, sock, addr, 0); if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] TCP link allocation failed"); return -1; } + link->active_users++; topo_node_apply_reality(link, reality); etcp_tcp_link_start_connect(link, addr, 0); - return 1; + int cancelled = link->closing || conn->close_requested; + etcp_link_release(link); + return cancelled ? -1 : 1; } /* Проверка совместимости и создание используют один обход пар, включая TCP fallback и IPv6. */ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni, - struct ETCP_SOCKET* specific_sock, struct UTUN_INSTANCE* check_inst) { + struct ETCP_SOCKET* specific_sock, int force_socket, struct UTUN_INSTANCE* check_inst) { if (!ni) { DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] link matching without node snapshot"); return 0; } if (!check_inst && ni->timestamp < entry->node_timestamp) { DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] ignore stale addresses node=%016llx current=%llu incoming=%llu", @@ -275,7 +281,7 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni, if (zero) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v4 skip zero-addr node=0x%016llx", (unsigned long long)nid); continue; } struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port); - struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin)); + struct sockaddr_storage sa = {0}; memcpy(&sa, &sin, sizeof(sin)); uint32_t tip; memcpy(&tip, a->addr, 4); const struct TOPO_SOCKMETA4* m = NULL; for (const struct TOPO_SOCKMETA4* mm = ni->v4_sock_meta; mm; mm = mm->next) @@ -295,28 +301,29 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni, const struct sock_view* v = &views[vi]; if (!v->is_tcp) continue; tcp_sockets++; - if (specific_sock ? v->sock != specific_sock : !sock_match(v, a->type, tcfg, profile, tip)) continue; + if ((specific_sock && v->sock != specific_sock) || + (!force_socket && !sock_match(v, a->type, tcfg, profile, tip))) continue; if (check_inst) return 1; int added = ncd_add_tcp_link(conn, v->sock, &sa, topo_node_find_reality_sock(ni, a->socket_id)); - if (added > 0) link_count += added; + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } if (!specific_sock && !tcp_sockets) { if (check_inst) return 1; int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id)); - if (added > 0) link_count += added; + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } } if (a->protocol & TOPO_PROTO_UDP) { - if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET) { + for (int vi = 0; vi < nviews; vi++) { + const struct sock_view* v = &views[vi]; + if (v->is_tcp || (specific_sock && v->sock != specific_sock) || + (!force_socket && !sock_match(v, a->type, tcfg, profile, tip))) continue; if (check_inst) return 1; - link_count += ncd_add_udp_link(conn, specific_sock, &sa); - } else if (!specific_sock) { - for (int vi = 0; vi < nviews; vi++) { - const struct sock_view* v = &views[vi]; - if (v->is_tcp || !sock_match(v, a->type, tcfg, profile, tip)) continue; - if (check_inst) return 1; - link_count += ncd_add_udp_link(conn, v->sock, &sa); - } + int added = ncd_add_udp_link(conn, v->sock, &sa); + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } } } @@ -344,39 +351,36 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni, (unsigned long long)nid, a->socket_id); continue; } int tcp_sockets = 0; - if (is_ll) { struct ETCP_SOCKET* sv = inst->etcp_sockets; - while (sv) { if (!sv->probe_only && sv->local_addr.ss_family == AF_INET6 && sv->netif_index) { sin6.sin6_scope_id = sv->netif_index; break; } sv = sv->next; } } - struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); + struct sockaddr_storage sa = {0}; memcpy(&sa, &sin6, sizeof(sin6)); for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) { if (!s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6) continue; tcp_sockets++; - if (specific_sock ? s != specific_sock : global && s->type == CFG_SERVER_TYPE_LOCAL) continue; + if ((specific_sock && s != specific_sock) || (is_ll && !s->netif_index) || + (!force_socket && target_class != SOCK_TC_PRIV && s->type == CFG_SERVER_TYPE_LOCAL)) continue; if (check_inst) return 1; if (is_ll) { sin6.sin6_scope_id = s->netif_index; memcpy(&sa, &sin6, sizeof(sin6)); } int added = ncd_add_tcp_link(conn, s, &sa, topo_node_find_reality_sock(ni, a->socket_id)); - if (added > 0) link_count += added; + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } if (!specific_sock && !tcp_sockets && !is_ll) { if (check_inst) return 1; int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id)); - if (added > 0) link_count += added; + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } } if (a->protocol & TOPO_PROTO_UDP) { - if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET6) { + for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) { + if (s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6 || + (specific_sock && s != specific_sock) || (is_ll && !s->netif_index) || + (!force_socket && target_class != SOCK_TC_PRIV && s->type == CFG_SERVER_TYPE_LOCAL)) continue; if (check_inst) return 1; - if (is_ll) sin6.sin6_scope_id = specific_sock->netif_index; - struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); - link_count += ncd_add_udp_link(conn, specific_sock, &sa); - } else if (!specific_sock) { - for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) { - if (s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6 || - (global && s->type == CFG_SERVER_TYPE_LOCAL)) continue; - if (check_inst) return 1; - if (is_ll) sin6.sin6_scope_id = s->netif_index; - struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); - link_count += ncd_add_udp_link(conn, s, &sa); - } + if (is_ll) sin6.sin6_scope_id = s->netif_index; + struct sockaddr_storage sa = {0}; memcpy(&sa, &sin6, sizeof(sin6)); + int added = ncd_add_udp_link(conn, s, &sa); + if (added < 0 || entry->closed || conn->close_requested) return -1; + link_count += added; } } } @@ -394,7 +398,7 @@ int node_conn_direct_compatible(struct UTUN_INSTANCE* inst, uint64_t node_id) { if (conn && conn->state != 2 && !conn->close_requested && conn->links_up) return 1; struct TOPO_NODE* node = topo_node_registry_find(inst->topo_groups, node_id); if (!node) { DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] compatibility unknown snapshot node=%016llx", (unsigned long long)node_id); return 0; } - if (ncd_create_links(NULL, node, NULL, inst) > 0) return 1; + if (ncd_create_links(NULL, node, NULL, 0, inst) > 0) return 1; /* Входящее подключение тоже допустимо: REVERSE_REQ просит пира открыть наш опубликованный адрес. * TCP имеет собственный исходящий socket fallback; для UDP нужен известный UDP-сокет пира. * Это возможность попытки, не подтверждение маршрута/работоспособности скрытых сокетов пира. */ @@ -448,7 +452,7 @@ int node_conn_direct_update_node(struct UTUN_INSTANCE* inst, uint64_t node_id) { struct ETCP_CONN* conn = entry->conn; if (etcp_conn_ref_take(conn) < 0) { topo_node_destroy(inst->topo_groups, copy); return -1; } entry->active_calls++; - int added = ncd_create_links(entry, copy, NULL, NULL); + int added = ncd_create_links(entry, copy, NULL, 0, NULL); DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] apply node=%016llx timestamp=%llu up=%d added=%d", (unsigned long long)node_id, (unsigned long long)copy->timestamp, entry->up, added); entry->active_calls--; ncd_collect(entry); @@ -818,7 +822,7 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id, ncd_callb entry->connect_timer = timer; entry->timed_out = 0; DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] connect deadline node=%016llx timeout_tb=%d", (unsigned long long)node_id, delay); } - if (copy && ncd_create_links(entry, copy, specific_sock, NULL) < 0) goto done; + if (copy && ncd_create_links(entry, copy, specific_sock, specific_sock != NULL, NULL) < 0) goto done; if (entry->closed || conn->close_requested || conn->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] open cancelled by transport callback node=%016llx", (unsigned long long)node_id); goto done; } @@ -979,135 +983,41 @@ struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h) { /* ═══════════ Управление линками на конкретном сокете (для auto_socket) ═══════════ */ -/* ncd_add_udp_link_random / ncd_add_tcp_link_random — собирают совместимые адреса пира, - * случайно выбирают один, проверяют что линк на (sock, addr, proto) ещё не существует, - * и создают новый. */ -static int ncd_add_udp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) { - int family = sock->local_addr.ss_family; - if (family == AF_INET) { - struct TOPO_ADDR4* compat[64]; int count = 0; - for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { - if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue; - int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; } - if (zero) continue; - if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a; - } - if (count == 0) return 0; - uint32_t r; random_bytes((uint8_t*)&r, sizeof(r)); - struct TOPO_ADDR4* a = compat[r % count]; - struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; - memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port); - struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin)); - if (etcp_link_find_by_addr(sock, &sa, 0)) return 0; - if (etcp_link_new(conn, sock, &sa, 0)) return 1; - return 0; - } - if (family == AF_INET6) { - struct TOPO_ADDR6* compat[64]; int count = 0; - for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) { - if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue; - int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; } - if (zero) continue; - if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a; - } - if (count == 0) return 0; - uint32_t r; random_bytes((uint8_t*)&r, sizeof(r)); - struct TOPO_ADDR6* a = compat[r % count]; - struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6; - memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port); - if ((a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index; - struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); - if (etcp_link_find_by_addr(sock, &sa, 0)) return 0; - if (etcp_link_new(conn, sock, &sa, 0)) return 1; - return 0; - } - return 0; -} - -static int ncd_add_tcp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) { - int family = sock->local_addr.ss_family; - if (family == AF_INET) { - struct TOPO_ADDR4* compat[64]; int count = 0; - for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { - if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue; - int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; } - if (zero) continue; - if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a; - } - if (count == 0) return 0; - uint32_t r; random_bytes((uint8_t*)&r, sizeof(r)); - struct TOPO_ADDR4* a = compat[r % count]; - struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; - memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port); - struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin)); - if (etcp_link_find_by_addr(sock, &sa, 1)) return 0; - struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0); - if (!tlink) return 0; - tlink->is_tcp = 1; - topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id)); - etcp_tcp_link_start_connect(tlink, &sa, a->port); - return 1; - } - if (family == AF_INET6) { - struct TOPO_ADDR6* compat[64]; int count = 0; - for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) { - if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue; - int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; } - if (zero) continue; - if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a; - } - if (count == 0) return 0; - uint32_t r; random_bytes((uint8_t*)&r, sizeof(r)); - struct TOPO_ADDR6* a = compat[r % count]; - struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6; - memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port); - if ((a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index; - struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); - if (etcp_link_find_by_addr(sock, &sa, 1)) return 0; - struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0); - if (!tlink) return 0; - tlink->is_tcp = 1; - topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id)); - etcp_tcp_link_start_connect(tlink, &sa, a->port); - return 1; - } - return 0; -} - +/* Снимок entry удерживает список через callbacks, которые могут закрыть соседних владельцев. */ int ncd_add_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) { - if (!inst || !sock) return 0; - int added = 0, total_conns = 0, skipped_closed = 0, skipped_no_addr = 0; - - struct ll_entry* qe = inst->connections ? inst->connections->head : NULL; - while (qe) { - struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data; - struct ETCP_CONN* conn = ce->conn; - qe = qe->next; - if (!conn) continue; - total_conns++; - - if (conn->state == 2) { skipped_closed++; continue; } - - struct TOPO_NODE* ni = ncd_lookup_node(inst, ce->peer_node_id); - if (!ni) { skipped_no_addr++; continue; } - - int n; - if (sock->is_tcp) n = ncd_add_tcp_link_random(conn, ni, sock); - else n = ncd_add_udp_link_random(conn, ni, sock); - if (n > 0) { - DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] add_socket_links sock=%s fd=%d conn=[%s] node=0x%016llx addr=%s is_tcp=%d", - sock->name, (int)sock->fd, conn->log_name, - (unsigned long long)ce->peer_node_id, - sockaddr_storage_to_str(&sock->local_addr).str, sock->is_tcp); - added += n; - } - topo_node_registry_unref(inst->topo_groups, ni->node_id); + if (!inst || !sock || sock->instance != inst || sock->probe_only || + (sock->local_addr.ss_family != AF_INET && sock->local_addr.ss_family != AF_INET6)) { + DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid added socket=%p", (void*)sock); return -1; } - - DEBUG_INFO(DEBUG_CATEGORY_NCD, - "[ncd] add_socket_links done: sock=%s is_tcp=%d total_conns=%d skipped(cl=%d,no_addr=%d) added=%d", - sock->name, sock->is_tcp, total_conns, skipped_closed, skipped_no_addr, added); - return added; + size_t count = 0; + for (struct ncd_entry* entry = inst->ncd_registry; entry; entry = entry->next) count++; + if (!count) return 0; + struct ncd_entry** entries = u_calloc(count, sizeof(*entries)); + if (!entries) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] added socket snapshot allocation failed"); return -1; } + size_t held = 0; + for (struct ncd_entry* entry = inst->ncd_registry; entry; entry = entry->next) { + entry->active_calls++; entries[held++] = entry; + } + int added = 0, failed = 0; + for (size_t i = 0; i < held; i++) { + struct ncd_entry* entry = entries[i]; + struct ETCP_CONN* conn = entry->conn; + if (!entry->closed && !conn->fin_wait && etcp_conn_ref_take(conn) == 0) { + struct TOPO_NODE* known = ncd_lookup_node(inst, entry->node_id); + struct TOPO_NODE* copy = known ? ncd_copy_node(inst, known) : NULL; + if (known) topo_node_registry_unref(inst->topo_groups, entry->node_id); + if (copy) { + int result = ncd_create_links(entry, copy, sock, 0, NULL); + if (result < 0) failed = 1; else added += result; + topo_node_destroy(inst->topo_groups, copy); + } else if (known) failed = 1; + etcp_conn_ref_free(conn); + } + entry->active_calls--; ncd_collect(entry); + } + u_free(entries); + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] added socket=%s transports=%zu links=%d failed=%d", sock->name, held, added, failed); + return failed ? -1 : added; } int ncd_remove_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {