jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix stale socket references in connectivity probes

Browse Source
This commit is contained in:
evgeny committed 2026-10-25 22:30:52 +01:00
1 parent 8b0002e489
commit e035f53ab6
4 files changed
+322 -47
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/route_connectivity.c
+104 -46
View File
@@ -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*)&target; 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*)&target; 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;
}
}
}
src/routing_layer/route_connectivity.h
+3 -1
View File
@@ -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);
tests/Makefile.am
+4
View File
@@ -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)
tests/test_route_connectivity.c
+211
View File
@@ -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;
}