diff --git a/src/routing_layer/route_connectivity.c b/src/routing_layer/route_connectivity.c index 896e7bef..62cdc12e 100644 --- a/src/routing_layer/route_connectivity.c +++ b/src/routing_layer/route_connectivity.c @@ -42,17 +42,17 @@ struct conn_probe_ctx { struct sockaddr_storage target_addr; uint8_t peer_pubkey[SC_PUBKEY_SIZE]; - struct ETCP_SOCKET* candidate_sockets[CONN_MAX_SOCKET_CANDIDATES]; + /* Снимок идентичностей: сокеты могут исчезнуть между асинхронными пингами. */ + struct { uint8_t id; uint64_t generation; } candidates[CONN_MAX_SOCKET_CANDIDATES]; uint8_t candidate_count; uint8_t candidate_index; + uint8_t invalidated; // часть кандидатов потеряна: отрицательный результат недостоверен - uint16_t best_across_sockets; // min RTT по всем сокетам uint8_t count_total; // 3 на серию (1 для TCP) uint8_t count_sent; uint8_t count_ok; uint16_t min_rtt; // min RTT в текущей серии uint16_t timeout_ms; - void* ping_timer; uint8_t reality_set; struct reality_client_config reality; struct stcp_client* tcp_cli; // хендл влётного TCP-пинга (для отмены), NULL для UDP @@ -61,7 +61,8 @@ struct conn_probe_ctx { // ---- forward ---- static void conn_probe_single_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len); -static void conn_probe_finish(struct conn_probe_ctx* ctx, int ok); +enum conn_probe_result { CONN_PROBE_FAILED, CONN_PROBE_REACHABLE, CONN_PROBE_INVALIDATED }; +static void conn_probe_finish(struct conn_probe_ctx* ctx, enum conn_probe_result result); static void conn_probe_start_series(struct conn_probe_ctx* ctx); // ---- helpers ---- @@ -147,36 +148,58 @@ static int conn_match_candidate_sockets(struct UTUN_INSTANCE* instance, // ---- probe lifecycle ---- +/* Заимствуем только живой UDP endpoint точного поколения, только на время текущего вызова uasync. */ +static struct ETCP_SOCKET* conn_probe_socket(const struct conn_probe_ctx* ctx) { + if (!ctx->candidates[ctx->candidate_index].generation) return NULL; + for (struct ETCP_SOCKET* sock = ctx->instance->etcp_sockets; sock; sock = sock->next) { + if (sock->sock_id != ctx->candidates[ctx->candidate_index].id || + sock->generation != ctx->candidates[ctx->candidate_index].generation) continue; + return !sock->is_tcp && !sock->probe_only && sock->fd != SOCKET_INVALID && + sock->local_addr.ss_family == ctx->target_addr.ss_family ? sock : NULL; + } + return NULL; +} + +/* Потеря локального endpoint не доказывает недоступность пира; старые RTT этой серии отбрасываем. */ +static void conn_probe_invalidate(struct conn_probe_ctx* ctx) { + DEBUG_WARN(DEBUG_CATEGORY_BGP, "probe candidate invalidated group=%016llx node=%016llx target=%s " + "socket=%u generation=%016llx candidate=%u/%u completed=%u/%u successful=%u", + (unsigned long long)ctx->group->group_id, (unsigned long long)ctx->nq->node_id, + sockaddr_storage_to_str(&ctx->target_addr).str, ctx->candidates[ctx->candidate_index].id, + (unsigned long long)ctx->candidates[ctx->candidate_index].generation, + ctx->candidate_index + 1, ctx->candidate_count, ctx->count_sent, ctx->count_total, ctx->count_ok); + ctx->invalidated = 1; +} + +/* 0 — пинг принят, 1 — кандидат устарел, отрицательное значение — ошибка отправки. */ static int conn_probe_send(struct conn_probe_ctx* ctx) { + DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "probe send group=%016llx node=%016llx target=%s tcp=%u candidate=%u/%u attempt=%u/%u", + (unsigned long long)ctx->group->group_id, (unsigned long long)ctx->nq->node_id, + sockaddr_storage_to_str(&ctx->target_addr).str, ctx->is_tcp, + ctx->candidate_index + 1, ctx->candidate_count, ctx->count_sent + 1, ctx->count_total); if (ctx->is_tcp) { return etcp_send_tcp_ping(ctx->instance, ctx->peer_pubkey, &ctx->target_addr, ctx->timeout_ms, conn_probe_single_cb, ctx, &ctx->tcp_cli, ctx->reality_set ? &ctx->reality : NULL); } - struct ETCP_SOCKET* sock = ctx->candidate_sockets[ctx->candidate_index]; + struct ETCP_SOCKET* sock = conn_probe_socket(ctx); + if (!sock) return 1; return etcp_send_ping_to_socket(ctx->instance, sock, ctx->peer_pubkey, &ctx->target_addr, ctx->timeout_ms, conn_probe_single_cb, ctx, NULL, 0, 0); } static void conn_probe_start_series(struct conn_probe_ctx* ctx) { - if (ctx->candidate_index >= ctx->candidate_count) { - // все сокеты перебраны - if (ctx->best_across_sockets != 65535) conn_probe_finish(ctx, 1); - else conn_probe_finish(ctx, 0); - return; - } - - ctx->count_sent = 0; - ctx->count_ok = 0; - ctx->min_rtt = 65535; - - int ret = conn_probe_send(ctx); - if (ret != 0) { - DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe: cannot start ping (tcp=%d)", ctx->is_tcp); - ctx->count_sent = ctx->count_total; // simulate full failure + while (ctx->candidate_index < ctx->candidate_count) { + ctx->count_sent = 0; ctx->count_ok = 0; ctx->min_rtt = UINT16_MAX; + int ret = conn_probe_send(ctx); + if (!ret) return; + if (ret == 1) conn_probe_invalidate(ctx); + else DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe cannot start target=%s tcp=%u candidate=%u/%u error=%d", + sockaddr_storage_to_str(&ctx->target_addr).str, ctx->is_tcp, + ctx->candidate_index + 1, ctx->candidate_count, ret); ctx->candidate_index++; - conn_probe_start_series(ctx); } + conn_probe_finish(ctx, ctx->invalidated ? CONN_PROBE_INVALIDATED : CONN_PROBE_FAILED); } static void conn_probe_single_cb(int success, uint16_t rtt, void* arg, @@ -184,20 +207,30 @@ static void conn_probe_single_cb(int success, uint16_t rtt, void* arg, (void)nonce; (void)resp_data; (void)resp_data_len; struct conn_probe_ctx* ctx = (struct conn_probe_ctx*)arg; if (!ctx) return; + if (!ctx->nq) { conn_probe_finish(ctx, CONN_PROBE_FAILED); return; } + if (!ctx->is_tcp && !conn_probe_socket(ctx)) { + conn_probe_invalidate(ctx); + ctx->candidate_index++; + conn_probe_start_series(ctx); + return; + } ctx->count_sent++; if (success) { ctx->count_ok++; if (rtt < ctx->min_rtt) ctx->min_rtt = rtt; } - DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "probe ping: ok=%d rtt=%u sent=%d/%d tcp=%d", - success, rtt, ctx->count_sent, ctx->count_total, ctx->is_tcp); + DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "probe result group=%016llx node=%016llx target=%s ok=%d rtt=%u completed=%u/%u tcp=%u", + (unsigned long long)ctx->group->group_id, (unsigned long long)ctx->nq->node_id, + sockaddr_storage_to_str(&ctx->target_addr).str, success, rtt, ctx->count_sent, ctx->count_total, ctx->is_tcp); if (ctx->count_sent < ctx->count_total) { int ret = conn_probe_send(ctx); if (ret != 0) { - DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe: cannot continue ping (tcp=%d)", ctx->is_tcp); + if (ret == 1) conn_probe_invalidate(ctx); + else DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe cannot continue target=%s tcp=%u error=%d", + sockaddr_storage_to_str(&ctx->target_addr).str, ctx->is_tcp, ret); ctx->count_sent = ctx->count_total; // force finish series ctx->count_ok = 0; } else return; // следующий пинг отправлен, ждём callback @@ -205,8 +238,7 @@ static void conn_probe_single_cb(int success, uint16_t rtt, void* arg, // серия пингов завершена if (ctx->count_ok > 0) { - if (ctx->min_rtt < ctx->best_across_sockets) ctx->best_across_sockets = ctx->min_rtt; - conn_probe_finish(ctx, 1); + conn_probe_finish(ctx, CONN_PROBE_REACHABLE); } else { // этот сокет не подошёл — пробуем следующий ctx->candidate_index++; @@ -214,42 +246,55 @@ static void conn_probe_single_cb(int success, uint16_t rtt, void* arg, } } -static void conn_probe_finish(struct conn_probe_ctx* ctx, int ok) { +static void conn_probe_finish(struct conn_probe_ctx* ctx, enum conn_probe_result result) { if (!ctx) return; + struct UTUN_INSTANCE* instance = ctx->instance; + struct TOPO_GROUP* group = ctx->group; + struct TOPO_GROUP_NODE* nq = ctx->nq; + int done = 0; if (ctx->nq) { struct TOPO_CONNECTIVITY* c = &ctx->nq->connectivity; struct conn_probe_ctx** p = (struct conn_probe_ctx**)&c->probe_list; while (*p) { if (*p == ctx) { *p = ctx->next; break; } p = &(*p)->next; } - switch (ctx->addr_type) { + int ok = result == CONN_PROBE_REACHABLE; + if (result != CONN_PROBE_INVALIDATED) switch (ctx->addr_type) { case TOPO_ADDR_INTERFACE: c->interface_status = ok ? PROBE_RESULT_REACHABLE : (c->interface_status == PROBE_RESULT_UNKNOWN ? PROBE_RESULT_UNREACHABLE : c->interface_status); - if (ok && ctx->best_across_sockets < c->interface_min_rtt) c->interface_min_rtt = ctx->best_across_sockets; + if (ok && ctx->min_rtt < c->interface_min_rtt) c->interface_min_rtt = ctx->min_rtt; c->interface_probe_time = get_time_tb(); c->real_status = ok ? PROBE_RESULT_REACHABLE : (c->real_status == PROBE_RESULT_UNKNOWN ? PROBE_RESULT_UNREACHABLE : c->real_status); - if (ok && ctx->best_across_sockets < c->real_min_rtt) c->real_min_rtt = ctx->best_across_sockets; + if (ok && ctx->min_rtt < c->real_min_rtt) c->real_min_rtt = ctx->min_rtt; c->real_probe_time = get_time_tb(); break; case TOPO_ADDR_NAT: c->nat_status = ok ? PROBE_RESULT_REACHABLE : (c->nat_status == PROBE_RESULT_UNKNOWN ? PROBE_RESULT_UNREACHABLE : c->nat_status); - if (ok && ctx->best_across_sockets < c->nat_min_rtt) c->nat_min_rtt = ctx->best_across_sockets; + if (ok && ctx->min_rtt < c->nat_min_rtt) c->nat_min_rtt = ctx->min_rtt; c->nat_probe_time = get_time_tb(); break; } if (c->pending_count > 0) c->pending_count--; + else DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe pending underflow group=%016llx node=%016llx", + (unsigned long long)group->group_id, (unsigned long long)nq->node_id); + DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "probe address complete group=%016llx node=%016llx target=%s result=%s remaining=%u", + (unsigned long long)group->group_id, (unsigned long long)nq->node_id, + sockaddr_storage_to_str(&ctx->target_addr).str, + ok ? "reachable" : result == CONN_PROBE_INVALIDATED ? "invalidated" : "unreachable", c->pending_count); if (c->pending_count == 0) { c->probe_status = PROBE_STATUS_DONE; - DEBUG_INFO(DEBUG_CATEGORY_BGP, "connectivity probe DONE for node %016llx: intf=%d nat=%d real=%d", - (unsigned long long)ctx->nq->node_id, + done = 1; + DEBUG_INFO(DEBUG_CATEGORY_BGP, "connectivity probe DONE group=%016llx node=%016llx: intf=%d nat=%d real=%d", + (unsigned long long)group->group_id, (unsigned long long)nq->node_id, c->interface_status, c->nat_status, c->real_status); - if (ctx->instance && ctx->instance->topo_sqlite_db) - topo_node_sqlite_nodeinfo_updated(ctx->instance->topo_sqlite_db, ctx->nq->node_id); - topo_fire_nodeinfo_cbk(ctx->instance, ctx->group, ctx->nq); } } u_free(ctx); + if (done) { + if (instance->topo_sqlite_db) topo_node_sqlite_nodeinfo_updated(instance->topo_sqlite_db, nq->node_id); + topo_fire_nodeinfo_cbk(instance, group, nq); + } } // ---- public API ---- @@ -292,7 +337,7 @@ static void route_connectivity_probe_deferred_cb(void* arg) { #endif void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* nq) { - if (!instance || !nq) return; + if (!instance || !group || !nq) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe: missing instance/group/node"); return; } struct TOPO_NODE* ni = topo_node_registry_find(instance->topo_groups, nq->node_id); if (!ni) return; if (nq->node_id == instance->node_id) return; // не пингуем себя @@ -362,10 +407,11 @@ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_G struct sockaddr_storage target; memset(&target, 0, sizeof(target)); struct sockaddr_in* sin = (struct sockaddr_in*)⌖ sin->sin_family = AF_INET; sin->sin_addr.s_addr = ip; sin->sin_port = htons(port); - struct conn_probe_ctx* ctx = u_calloc(1, sizeof(struct conn_probe_ctx)); if (!ctx) continue; + struct conn_probe_ctx* ctx = u_calloc(1, sizeof(struct conn_probe_ctx)); + if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe IPv4 context allocation failed"); continue; } ctx->instance = instance; ctx->group = group; ctx->nq = nq; ctx->addr_type = a->type; ctx->target_addr = target; memcpy(ctx->peer_pubkey, ni->public_key, SC_PUBKEY_SIZE); - ctx->timeout_ms = g_probe_timeout_ms; ctx->best_across_sockets = 65535; + ctx->timeout_ms = g_probe_timeout_ms; if (a->protocol & TOPO_PROTO_TCP) { const struct TOPO_REALITY_SOCK *rs = topo_node_find_reality_sock(ni, a->socket_id); @@ -386,11 +432,13 @@ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_G struct ETCP_SOCKET* candidates[CONN_MAX_SOCKET_CANDIDATES]; int cand_count = conn_match_candidate_sockets(instance, a->type, tcfg, profile, &target, candidates, CONN_MAX_SOCKET_CANDIDATES); if (cand_count == 0) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "probe: no local sockets for target %s:%u type=%d", ip_to_str(&ip, AF_INET).str, port, a->type); u_free(ctx); continue; } - memcpy(ctx->candidate_sockets, candidates, cand_count * sizeof(struct ETCP_SOCKET*)); + for (int i = 0; i < cand_count; i++) { + ctx->candidates[i].id = candidates[i]->sock_id; + ctx->candidates[i].generation = candidates[i]->generation; + } ctx->candidate_count = cand_count; ctx->candidate_index = 0; ctx->count_total = CONN_PROBE_COUNT; } ctx->next = (struct conn_probe_ctx*)nq->connectivity.probe_list; nq->connectivity.probe_list = ctx; pend++; - conn_probe_start_series(ctx); } } @@ -409,10 +457,11 @@ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_G struct sockaddr_in6* sin6 = (struct sockaddr_in6*)⌖ sin6->sin6_family = AF_INET6; memcpy(&sin6->sin6_addr, a6->addr, 16); sin6->sin6_port = htons(a6->port); - struct conn_probe_ctx* ctx = u_calloc(1, sizeof(struct conn_probe_ctx)); if (!ctx) continue; - ctx->instance = instance; ctx->nq = nq; ctx->addr_type = a6->type; ctx->target_addr = target; + struct conn_probe_ctx* ctx = u_calloc(1, sizeof(struct conn_probe_ctx)); + if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "probe IPv6 context allocation failed"); continue; } + ctx->instance = instance; ctx->group = group; ctx->nq = nq; ctx->addr_type = a6->type; ctx->target_addr = target; memcpy(ctx->peer_pubkey, ni->public_key, SC_PUBKEY_SIZE); - ctx->timeout_ms = g_probe_timeout_ms; ctx->best_across_sockets = 65535; + ctx->timeout_ms = g_probe_timeout_ms; if (a6->protocol & TOPO_PROTO_TCP) { const struct TOPO_REALITY_SOCK *rs = topo_node_find_reality_sock(ni, a6->socket_id); @@ -430,11 +479,13 @@ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_G int cand_count = conn_match_candidate_sockets(instance, a6->type, CFG_SERVER_TYPE_UNKNOWN, NULL, &target, candidates, CONN_MAX_SOCKET_CANDIDATES); if (cand_count == 0) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "probe: no local sockets for IPv6 target type=%d", a6->type); u_free(ctx); continue; } - memcpy(ctx->candidate_sockets, candidates, cand_count * sizeof(struct ETCP_SOCKET*)); + for (int i = 0; i < cand_count; i++) { + ctx->candidates[i].id = candidates[i]->sock_id; + ctx->candidates[i].generation = candidates[i]->generation; + } ctx->candidate_count = cand_count; ctx->candidate_index = 0; ctx->count_total = CONN_PROBE_COUNT; } ctx->next = (struct conn_probe_ctx*)nq->connectivity.probe_list; nq->connectivity.probe_list = ctx; pend++; - conn_probe_start_series(ctx); } } @@ -447,6 +498,13 @@ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_G nq->connectivity.pending_count = pend; DEBUG_INFO(DEBUG_CATEGORY_BGP, "connectivity probe started for node %016llx: %d series pending", (unsigned long long)ni->node_id, pend); + /* Все контексты учтены до отправки: отказ может синхронно завершить и освободить текущий. */ + struct conn_probe_ctx* ctx = nq->connectivity.probe_list; + while (ctx) { + struct conn_probe_ctx* next = ctx->next; + conn_probe_start_series(ctx); + ctx = next; + } } } diff --git a/src/routing_layer/route_connectivity.h b/src/routing_layer/route_connectivity.h index 2e7d3693..3f999d0c 100644 --- a/src/routing_layer/route_connectivity.h +++ b/src/routing_layer/route_connectivity.h @@ -20,7 +20,9 @@ struct TOPO_GROUP; * UDP: до 8 сокетов, серия CONN_PROBE_COUNT с каждого до первого успешного сокета; * TCP: один STCP ping с REALITY-параметрами при наличии. RTT — 0.1 мс, UINT16_MAX неизвестен. * До 16 уникальных IP:port на семейство. Собственный/спящий узел и текущие пробы пропускаются; - * NAT/standby могут отложить старт. Результат — в nq->connectivity, завершение уведомляет topo callback. */ + * NAT/standby могут отложить старт. UDP-кандидат хранится как ID+поколение: исчезновение endpoint + * отменяет его серию без отрицательного результата; новые сокеты попадут в следующий запуск. + * Результат — в nq->connectivity, завершение уведомляет topo callback (может удалить nq). */ void route_connectivity_probe_node(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* nq); diff --git a/tests/Makefile.am b/tests/Makefile.am index 519499a5..63c8b7ae 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -272,6 +272,10 @@ check_PROGRAMS += test_local_transport test_local_transport_SOURCES = test_local_transport.c $(test_clock_sources) test_local_transport_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_local_transport_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=local_interfaces -Wl,--wrap=socket_sendto +check_PROGRAMS += test_route_connectivity +test_route_connectivity_SOURCES = test_route_connectivity.c $(test_clock_sources) +test_route_connectivity_LDADD = $(LIBUTUN) $(COMMON_LIBS) +test_route_connectivity_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_send_ping_to_socket -Wl,--wrap=socket_sendto check_PROGRAMS += test_local_routes test_local_routes_SOURCES = test_local_routes.c $(test_clock_sources) test_local_routes_LDADD = $(LIBUTUN) $(COMMON_LIBS) diff --git a/tests/test_route_connectivity.c b/tests/test_route_connectivity.c new file mode 100644 index 00000000..67bfb640 --- /dev/null +++ b/tests/test_route_connectivity.c @@ -0,0 +1,211 @@ +/* Настоящие UDP/PONG, таймеры и освобождение сокетов; потери управляются на границе sendto. + * Проверяем исчезновение endpoint, смену поколения, учёт синхронных отказов и teardown. */ +#include "nat_fixture.h" +#include "test_clock.h" +#include "routing_layer/route_connectivity.h" +#include "../lib/memory_pool.h" + +static struct UTUN_INSTANCE* owner; +static struct TOPO_GROUP* group; +static struct TOPO_GROUP_NODE* node; +static struct ETCP_SOCKET* first_socket; +static socket_t drop_fd; +static int sends, notifications, fail_send, drop_after_first, drop_first, remove_on_complete; + +int __real_etcp_send_ping_to_socket(struct UTUN_INSTANCE*, struct ETCP_SOCKET*, const uint8_t*, + const struct sockaddr_storage*, int, etcp_ping_callback_t, void*, const uint8_t*, size_t, uint8_t); + +/* Наблюдаем только пробы route_connectivity; аварийный отказ не создаёт pending PING. */ +int __wrap_etcp_send_ping_to_socket(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* socket, const uint8_t* key, + const struct sockaddr_storage* address, int timeout, etcp_ping_callback_t callback, void* arg, + const uint8_t* data, size_t size, uint8_t flags) { + if (inst == owner && !size && !flags) { + sends++; + if (!first_socket) first_socket = socket; + if (drop_first) { drop_fd = socket->fd; drop_first = 0; } + if (fail_send) { + DEBUG_WARN(DEBUG_CATEGORY_BGP, "test: synchronous UDP send rejection target=%s", sockaddr_storage_to_str(address).str); + return -9; + } + } + return __real_etcp_send_ping_to_socket(inst, socket, key, address, timeout, callback, arg, data, size, flags); +} + +ssize_t __real_socket_sendto(socket_t, const void*, size_t, const struct sockaddr*, socklen_t); +ssize_t __wrap_socket_sendto(socket_t fd, const void* data, size_t size, const struct sockaddr* address, socklen_t address_size) { + if (fd == drop_fd || (drop_after_first && sends >= 2 && first_socket && fd == first_socket->fd)) return (ssize_t)size; + return __real_socket_sendto(fd, data, size, address, address_size); +} + +static void completed(struct TOPO_GROUP* notified_group, struct TOPO_GROUP_NODE* notified_node, void* arg) { + (void)arg; + if (notified_node != node || node->connectivity.probe_status != PROBE_STATUS_DONE) return; + assert(notified_group == group && notified_node == node); + assert(node->connectivity.probe_status == PROBE_STATUS_DONE && !node->connectivity.pending_count); + assert(!node->connectivity.probe_list); + notifications++; + if (remove_on_complete) { topo_nodeq_remove_node(group, node); node = NULL; } +} + +/* Отдельный registry-снимок пира с реальными bound-портами; группа владеет единственной ссылкой. */ +static void add_peer_address(struct TOPO_NODE* identity, struct UTUN_INSTANCE* peer, const char* name) { + struct ETCP_SOCKET* socket = test_etcp_bound_socket(peer, name); + if (socket->local_addr.ss_family == AF_INET) { + struct TOPO_ADDR4* address = memory_pool_alloc(owner->topo_groups->v4_addr_pool); assert(address); + *address = (struct TOPO_ADDR4){.next = identity->v4_addrs, .port = ss_get_port(&socket->local_addr), + .type = TOPO_ADDR_INTERFACE, .protocol = socket->is_tcp ? TOPO_PROTO_TCP : TOPO_PROTO_UDP}; + memcpy(address->addr, &((struct sockaddr_in*)&socket->local_addr)->sin_addr, 4); + identity->v4_addrs = address; + } else { + struct TOPO_ADDR6* address = memory_pool_alloc(owner->topo_groups->v6_addr_pool); assert(address); + *address = (struct TOPO_ADDR6){.next = identity->v6_addrs, .port = ss_get_port(&socket->local_addr), + .type = TOPO_ADDR_INTERFACE, .protocol = TOPO_PROTO_UDP}; + memcpy(address->addr, &((struct sockaddr_in6*)&socket->local_addr)->sin6_addr, 16); + identity->v6_addrs = address; + } +} + +static void setup(struct nat_fixture* fixture, int two_candidates, int mixed, int tcp) { + sends = notifications = fail_send = drop_after_first = drop_first = remove_on_complete = 0; + first_socket = NULL; drop_fd = SOCKET_INVALID; + nat_fixture_init(fixture); + owner = nat_fixture_node(fixture, "127.0.0.2", NULL, two_candidates ? + "[server:v6]\naddr=[::1]:0\ntype=public\n[server:extra]\naddr=[::1]:0\ntype=public\n" : + "[server:v6]\naddr=[::1]:0\ntype=public\n"); + struct UTUN_INSTANCE* peer = nat_fixture_node(fixture, "127.0.0.3", NULL, + "[server:v6]\naddr=[::1]:0\ntype=public\n[server:tcp]\naddr=127.0.0.3:0\ntype=public\ntransport=tcp\n"); + nat_detection_set_automatic(owner->nat_det, 0); nat_detection_set_automatic(peer->nat_det, 0); + group = topo_groups_find(owner->topo_groups, TOPO_GROUP_UTUN); assert(group); + struct TOPO_NODE* identity = u_calloc(1, sizeof(*identity)); assert(identity); + identity->node_id = peer->node_id; + memcpy(identity->public_key, peer->my_keys.public_key, 32); + memcpy(identity->ed25519_public_key, peer->my_ed25519_pubkey, 32); + if (tcp) add_peer_address(identity, peer, "tcp"); + else { + add_peer_address(identity, peer, "v6"); + if (mixed) add_peer_address(identity, peer, "udp"); + } + assert(topo_node_registry_store(owner->topo_groups, identity) == identity); + struct ll_entry* entry = queue_entry_new(sizeof(*node)); assert(entry); + node = (struct TOPO_GROUP_NODE*)entry; node->node_id = peer->node_id; + queue_data_put_with_index(group->nodes, entry); + utun_add_nodeinfo_cbk(owner, completed, NULL); +} + +static void start(void) { + route_connectivity_probe_node(owner, group, node); + assert(node->connectivity.pending_count && node->connectivity.probe_status == PROBE_STATUS_IN_PROGRESS); +} + +static void wait_done(struct nat_fixture* fixture) { + uint64_t deadline = get_time_tb() + 10000; + while (!notifications && get_time_tb() < deadline) test_clock_poll(fixture->ua, 10); + assert(notifications == 1 && !owner->pending_pings); + assert(node->connectivity.probe_status == PROBE_STATUS_DONE && !node->connectivity.pending_count); +} + +static void cleanup(struct nat_fixture* fixture) { + first_socket = NULL; drop_fd = SOCKET_INVALID; drop_after_first = 0; + nat_fixture_cleanup(fixture); owner = NULL; group = NULL; node = NULL; +} + +/* ID переиспользуется, но новый endpoint не должен получить пинг из старого снимка. */ +static void test_replaced_socket(int reuse_id, int partial_success) { + struct nat_fixture fixture; setup(&fixture, 0, 0, 0); + if (partial_success) drop_after_first = 1; + else drop_fd = test_etcp_bound_socket(owner, "v6")->fd; + start(); + if (partial_success) { + uint64_t deadline = get_time_tb() + 1000; + while (sends < 2 && get_time_tb() < deadline) test_clock_poll(fixture.ua, 1); + assert(sends == 2 && owner->pending_pings); + } + uint8_t id = first_socket->sock_id; + uint64_t generation = first_socket->generation; + first_socket = NULL; + etcp_socket_remove(test_etcp_bound_socket(owner, "v6")); + if (reuse_id) { + owner->next_socket_id = id; + struct CFG_SERVER server = {0}; snprintf(server.name, sizeof(server.name), "replacement"); + server.type = CFG_SERVER_TYPE_PUBLIC; + struct sockaddr_in6* address = (struct sockaddr_in6*)&server.ip; + address->sin6_family = AF_INET6; assert(inet_pton(AF_INET6, "::1", &address->sin6_addr) == 1); + struct ETCP_SOCKET* replacement = etcp_socket_add(owner, &server); assert(replacement); + assert(replacement->sock_id == id && replacement->generation != generation); + } + wait_done(&fixture); + assert(sends == (partial_success ? 2 : 1)); + assert(node->connectivity.interface_status == PROBE_RESULT_UNKNOWN && node->connectivity.real_status == PROBE_RESULT_UNKNOWN); + /* Общий PONG отдельно обновляет исторический RTT узла; отменённая серия не публикует результат/время пробы. */ + assert(!node->connectivity.interface_probe_time); + assert(partial_success ? node->connectivity.interface_min_rtt < UINT16_MAX : + node->connectivity.interface_min_rtt == UINT16_MAX); + cleanup(&fixture); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS probe socket removal reuse_id=%d partial_success=%d: UNKNOWN, no stale sends/probe result", + reuse_id, partial_success); +} + +static void test_changed_generation(int fallback) { + struct nat_fixture fixture; setup(&fixture, fallback, 0, 0); + drop_first = 1; + start(); assert(first_socket && first_socket->fd == drop_fd); + first_socket->generation++; // Та же память/ID, другая эпоха endpoint. + wait_done(&fixture); + assert(sends == (fallback ? 4 : 1)); + assert(node->connectivity.interface_status == (fallback ? PROBE_RESULT_REACHABLE : PROBE_RESULT_UNKNOWN)); + cleanup(&fixture); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS probe in-place generation change fallback=%d", fallback); +} + +/* Удаляется ещё не использованный кандидат; проверка текущего даёт обычные таймауты. */ +static void test_future_candidate(void) { + struct nat_fixture fixture; setup(&fixture, 1, 0, 0); + drop_first = 1; start(); + etcp_socket_remove(test_etcp_bound_socket(owner, !strcmp(first_socket->name, "v6") ? "extra" : "v6")); + wait_done(&fixture); + assert(sends == CONN_PROBE_COUNT && node->connectivity.interface_status == PROBE_RESULT_UNKNOWN); + cleanup(&fixture); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS probe removed future candidate: incomplete failure stays UNKNOWN"); +} + +static void test_immediate_failure(int remove_node) { + struct nat_fixture fixture; setup(&fixture, 0, 1, 0); + fail_send = 1; remove_on_complete = remove_node; + route_connectivity_probe_node(owner, group, node); + assert(sends == 2 && notifications == 1 && !owner->pending_pings); + if (remove_node) assert(!node); + else { + assert(!node->connectivity.pending_count && !node->connectivity.probe_list); + assert(node->connectivity.probe_status == PROBE_STATUS_DONE); + assert(node->connectivity.interface_status == PROBE_RESULT_UNREACHABLE); + } + cleanup(&fixture); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS mixed IPv4/IPv6 synchronous failures: one completion remove_node=%d", remove_node); +} + +static void test_cancel(int tcp, int destroy_instance) { + struct nat_fixture fixture; setup(&fixture, 0, 0, tcp); + if (!tcp) drop_fd = test_etcp_bound_socket(owner, "v6")->fd; + start(); + if (!destroy_instance) { + topo_nodeq_remove_node(group, node); node = NULL; + test_clock_run_for(fixture.ua, 1000); + assert(!notifications && !owner->pending_pings); + } + cleanup(&fixture); + assert(!notifications); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS probe cancellation tcp=%d instance_destroy=%d: timers/contexts released", tcp, destroy_instance); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); + debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO); + test_clock_init(1000); route_connectivity_set_probe_timeout_ms(10); + test_replaced_socket(0, 0); test_replaced_socket(1, 0); test_replaced_socket(1, 1); + test_changed_generation(0); test_changed_generation(1); test_future_candidate(); + test_immediate_failure(0); test_immediate_failure(1); + test_cancel(0, 0); test_cancel(0, 1); test_cancel(1, 0); test_cancel(1, 1); + route_connectivity_set_probe_timeout_ms(CONN_PROBE_TIMEOUT_MS); test_clock_finish(); + return 0; +}