diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index 61578e7b..76614f04 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -1184,6 +1184,9 @@ static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dea void etcp_link_close(struct ETCP_LINK* link) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); if (!link) return; + if (link->rx_timeout_timer) { + uasync_cancel_timeout(link->etcp->instance->ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL; + } callring_put(CR_LINK_CLOSE_IN, (uintptr_t)link, (uintptr_t)(link->etcp)); if (link->etcp && link->etcp->last_rr_link == link) link->etcp->last_rr_link = NULL; @@ -2607,6 +2610,7 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt, link->last_recv_timestamp=pkt->timestamp; link->last_recv_updated=1; link->total_decrypted += pkt->data_len; + if (pkt->data[0] != ETCP_KEEPALIVE) etcp_keepalive_watch_rx(link); uint8_t pkt_code = pkt->data[0]; @@ -2902,6 +2906,7 @@ int init_connections(struct UTUN_INSTANCE* instance) { } } + node_conn_direct_init(instance); etcp_keepalive_register(instance); // Return 1 if there was a partial socket initialization error diff --git a/src/transport_layer/node_conn_direct.c b/src/transport_layer/node_conn_direct.c index df073e1e..3f0bf8d3 100644 --- a/src/transport_layer/node_conn_direct.c +++ b/src/transport_layer/node_conn_direct.c @@ -26,6 +26,7 @@ #include "../lib/socket_compat.h" #include "../lib/platform_compat.h" #include +#include #define DEBUG_CATEGORY_NCD DEBUG_CATEGORY_ETCP @@ -649,11 +650,12 @@ static void ncd_fin_wait_timeout_cb(void* arg) { * * KEEP_ALIVE (пир просит не закрывать conn — у него ещё есть handle'ы): * - Если мы в fin_wait — отменяем его, conn продолжает работать. - * - Если у нас handle'ов нет — перезапускаем fin_wait (ждём ещё 5 сек). + * - Отсутствие наших handle'ов не отменяет потребность пира: FIN_WAIT снимается полностью. */ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) { if (!conn || !e || e->len < NCD_CONTROL_MSG_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%u", (void*)conn, e ? e->len : 0); if (e) { queue_dgram_free(e); queue_entry_free(e); } return; } @@ -676,13 +678,16 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) if (entry->handle_count > 0) { DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] refusing CLOSE, still have %d handles for 0x%016llx", entry->handle_count, (unsigned long long)sender_id); - { struct ncd_control_msg resp; resp.cmd = ETCP_RT_ID_NCD_CONTROL; resp.subcmd = NCD_SUBCMD_KEEP_ALIVE; - resp.node_id = conn->instance->node_id; - struct ll_entry* qe = queue_entry_new(0); - if (qe) { qe->dgram = u_malloc(sizeof(resp)); - if (qe->dgram) { memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp); - if (etcp_send(conn, qe) != 0) { queue_dgram_free(qe); queue_entry_free(qe); } } - else queue_entry_free(qe); } + { struct ncd_control_msg resp = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id }; + struct ll_entry* qe = ll_alloc_lldgram(sizeof(resp)); + if (!qe) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE allocation failed"); + else { + memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp); + if (etcp_send(conn, qe) != 0) { + DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE send failed node=%016llx", (unsigned long long)sender_id); + queue_dgram_free(qe); queue_entry_free(qe); + } + } } } else { DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] deferred close entry for CLOSE from 0x%016llx", (unsigned long long)sender_id); @@ -694,15 +699,12 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) if (entry && entry->conn->fin_wait) { entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->conn->fin_wait_clear_arg = NULL; if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; } - if (entry->handle_count <= 0) { - entry->conn->fin_wait = 1; - entry->fin_wait_timer = uasync_set_timeout(entry->ua, NCD_FIN_WAIT_TIMEOUT_TB, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait"); - DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE but no handles, restart fin_wait"); - } else { - DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive"); - } + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive handles=%d", entry->handle_count); } break; + default: + DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] unknown control subcmd=%u node=%016llx", msg->subcmd, (unsigned long long)sender_id); + break; } queue_dgram_free(e); queue_entry_free(e); } @@ -710,15 +712,19 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) /* ═══════════ инициализация глобального обработчика ═══════════ */ /* * Однократная привязка обработчика NCD control-сообщений к ETCP (etcp_bind). - * Вызывается при первом node_conn_direct_open, повторные вызовы безвредны + * Вызывается при запуске сети и node_conn_direct_open, повторные вызовы безвредны * (защита через inst->ncd_control_bound). */ -static void ncd_init_control_binding(struct UTUN_INSTANCE* inst) { - if (inst->ncd_control_bound) return; - etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler); +int node_conn_direct_init(struct UTUN_INSTANCE* inst) { + if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] init without instance"); return -1; } + if (inst->ncd_control_bound) return 0; + if (etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot bind CLOSE/KEEP_ALIVE handler"); return -1; + } inst->ncd_control_bound = 1; DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_RT_ID_NCD_CONTROL handler"); + return 0; } /* ═══════════ оживление «зависшего» (timed_out) entry ═══════════ */ @@ -776,7 +782,7 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id, struct ETCP_SOCKET* specific_sock) { if (!inst || !out_handle) return NCD_ERR; *out_handle = NULL; - ncd_init_control_binding(inst); + if (node_conn_direct_init(inst) < 0) return NCD_ERR; /* 1. Ищем в реестре — conn уже есть */ struct ncd_entry* entry = ncd_registry_find(inst, node_id); @@ -938,7 +944,7 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id, struct ETCP_SOCKET* specific_sock) { if (!inst || !out_handle || !ni) return NCD_ERR; *out_handle = NULL; - ncd_init_control_binding(inst); + if (node_conn_direct_init(inst) < 0) return NCD_ERR; /* 1. Ищем в реестре — conn уже есть */ { struct ncd_entry* entry = ncd_registry_find(inst, node_id); @@ -1134,9 +1140,14 @@ static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) { (unsigned long long)entry->node_id, entry->handle_count, force); if (entry->handle_count || entry->closed) { ncd_collect(entry); return; } struct ETCP_CONN* conn = entry->conn; - if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out || !conn->links_up) { + if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out) { ncd_finish(entry, 1); return; } + if (!conn->links_up) { + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] last handle released while DOWN, retaining until ETCP timeout node=%016llx", + (unsigned long long)entry->node_id); + ncd_finish(entry, 0); return; + } if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; } conn->fin_wait = 1; conn->fin_wait_clear_cb = ncd_fin_wait_cancelled; conn->fin_wait_clear_arg = entry; struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id }; @@ -1148,8 +1159,18 @@ static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) { DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] CLOSE send failed"); queue_dgram_free(e); queue_entry_free(e); } } - entry->fin_wait_timer = uasync_set_timeout(entry->ua, NCD_FIN_WAIT_TIMEOUT_TB, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait"); - DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx", (unsigned long long)entry->node_id); + uint64_t now = get_time_tb(), wait_tb = NCD_FIN_WAIT_TIMEOUT_TB; + for (struct ETCP_LINK* link = conn->links; link; link = link->next) { + if (!link->link_status || (!link->ka_my_sleeping && !link->ka_peer_sleeping)) continue; + if (link->rx_deadline_tb > now && link->rx_deadline_tb - now > wait_tb) wait_tb = link->rx_deadline_tb - now; + } + int delay = wait_tb > INT_MAX ? INT_MAX : (int)wait_tb; + entry->fin_wait_timer = uasync_set_timeout(entry->ua, delay, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait"); + if (!entry->fin_wait_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot schedule FIN_WAIT node=%016llx", (unsigned long long)entry->node_id); + ncd_finish(entry, 1); return; + } + DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx timeout_ms=%d", (unsigned long long)entry->node_id, delay / 10); ncd_collect(entry); } void node_conn_direct_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 0); } diff --git a/src/transport_layer/node_conn_direct.h b/src/transport_layer/node_conn_direct.h index 3cb3d79d..33d63c42 100644 --- a/src/transport_layer/node_conn_direct.h +++ b/src/transport_layer/node_conn_direct.h @@ -13,8 +13,8 @@ * * Модуль управляет ETCP-соединениями через непрозрачные handle'ы (NODE_CONN_DIRECT). * Каждый владелец открывает свой handle; несколько handle разделяют ETCP_CONN. - * close снимает только своё владение. Последний handle закрывает неготовый транспорт - * сразу; для UP начинает CLOSE/KEEP_ALIVE (до 5 s), пир может сохранить соединение. + * close снимает только своё владение. Последний handle работающего транспорта начинает + * CLOSE/KEEP_ALIVE (5 s либо срок пробуждения пира); DOWN удаляется по таймауту ETCP. * * Кто использует: * - conn_mgr (Connection Manager) — временные handle'ы попыток установления прямого транспорта @@ -36,6 +36,9 @@ struct ETCP_CONN; struct TOPO_NODE; struct ETCP_SOCKET; +/* Привязать CLOSE/KEEP_ALIVE при запуске ядра, в том числе без исходящих handles. */ +int node_conn_direct_init(struct UTUN_INSTANCE* inst); + /* ─── Return codes ─── */ #define NCD_NEW 0 /* создано новое соединение */ #define NCD_REUSED 1 /* переиспользовано существующее */