// etcp_router.c — Сервисный слой маршрутизации поверх ETCP // Упрощённый TCP: восстановление порядка (recv_q), дедупликация, без переповторов // ACK: периодическая отправка rx_seq (не чаще 100ms), idle-таймер для последнего seq // Inflight контроль через send_q: при переполнении — очередь + retry-таймер, без дропов /* /----- ack ack <-- id |(pause) | in ---> {Q} -> [src, etcp] --> ... --> [dst,etcp] -> {asm_q} -> {buf q} ---> out */ #include "etcp_router.h" #include "etcp.h" #include "utun_instance.h" #include "topo_group.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../lib/ll_queue.h" #include "../lib/u_async.h" #include // ==================================================================== // Внутренние forward declarations // ==================================================================== static void router_ack_timer_cb(void* arg); static void router_idle_ack_timer_cb(void* arg); static void router_send_ack(struct ETCP_ROUTER_CONN* rconn); static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn); static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn); static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, const uint8_t* payload, size_t payload_len); static int router_send_one(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len); static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, uint8_t flag_bits, int is_signed); static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t svc_id, uint8_t flag_bits); static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn); static void router_send_resume_cb(void* arg); static void router_no_route_retry_cb(void* arg); static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn); static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn); static void router_close_finalize(void* arg); static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr, size_t* pl_len); static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts); static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr); static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr); static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* pl, size_t pl_len); static void router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf); static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn); static void router_retrans_timer_cb(void* arg); static void router_incoming_q_cb(struct ll_queue* q, void* arg); static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn); static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn); static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn); static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn); // ==================================================================== // Транзитные очереди — per (src, dst) pair, backpressure через waiter на send_input_q static struct TRANSIT_QUEUE* transit_queue_find(struct ETCP_CONN* conn, uint64_t src_node_id, uint64_t dst_node_id) { if (!conn->transit_queues) return NULL; uint8_t key[16]; memcpy(key, &src_node_id, 8); memcpy(key + 8, &dst_node_id, 8); return (struct TRANSIT_QUEUE*)queue_find_data_by_index(conn->transit_queues, key); } static struct TRANSIT_QUEUE* transit_queue_get_or_create(struct ETCP_CONN* conn, uint64_t src_node_id, uint64_t dst_node_id) { if (!conn->transit_queues) { struct UASYNC* ua = conn->instance ? conn->instance->ua : NULL; if (!ua) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: no ua"); return NULL; } conn->transit_queues = queue_new(ua, TRANSIT_QUEUE_HASH, 0, 16, "transit_q_reg"); if (!conn->transit_queues) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_new failed"); return NULL; } } struct TRANSIT_QUEUE* tq = transit_queue_find(conn, src_node_id, dst_node_id); if (tq) return tq; size_t data_size = sizeof(struct TRANSIT_QUEUE) - sizeof(struct ll_entry); tq = (struct TRANSIT_QUEUE*)queue_entry_new(data_size); if (!tq) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_entry_new failed"); return NULL; } memset(&tq->waiter, 0, sizeof(tq->waiter)); tq->conn = conn; tq->q = queue_new(conn->instance->ua, 0, 0, 0, "transit_pair_q"); if (!tq->q) { queue_entry_free(&tq->ll); return NULL; } memcpy(tq->ll.data, &src_node_id, 8); memcpy(tq->ll.data + 8, &dst_node_id, 8); queue_data_put_with_index(conn->transit_queues, &tq->ll); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "transit_q: new src=%016llx dst=%016llx", (unsigned long long)src_node_id, (unsigned long long)dst_node_id); return tq; } static void transit_queue_destroy(struct ETCP_CONN* conn, struct TRANSIT_QUEUE* tq) { if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); struct ll_entry* e; while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); queue_free(tq->q); queue_remove_data(conn->transit_queues, &tq->ll); queue_entry_free(&tq->ll); } static void transit_queue_drain_cb(struct ll_queue* q, void* arg) { struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)arg; struct ETCP_CONN* conn = tq->conn; struct ll_entry* e = queue_data_get(tq->q); if (!e) { transit_queue_destroy(conn, tq); return; } int ret = etcp_send(conn, e); if (ret != 0) free_entry(e); if (queue_entry_count(tq->q) > 0) queue_waiter_wait(conn->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); else transit_queue_destroy(conn, tq); } void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn) { if (!conn || !conn->transit_queues) return; struct ll_entry* entry; while ((entry = queue_data_get(conn->transit_queues)) != NULL) { struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)entry; if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); struct ll_entry* e; while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); queue_free(tq->q); queue_entry_free(&tq->ll); } queue_free(conn->transit_queues); conn->transit_queues = NULL; } // ==================================================================== // Управление ROUTER_CONN // ==================================================================== static struct ETCP_ROUTER_CONN* router_conn_find(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) { if (!inst || !inst->router_conns) return NULL; uint8_t key[9]; memcpy(key, &remote_node_id, 8); key[8] = svc_id; return (struct ETCP_ROUTER_CONN*)queue_find_data_by_index(inst->router_conns, key); } // ==================================================================== // Отправка: router_send_one + send_q // ==================================================================== static int router_send_one(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) { return router_send_one_flags(rconn, payload, pl_len, 0, 0); } static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, uint8_t flag_bits, int is_signed) { struct UTUN_INSTANCE* inst = rconn->inst; uint32_t seq = flag_bits ? 0 : rconn->tx_seq++; size_t sig_len = is_signed ? SC_SIGN_SIZE : 0; size_t total_len = SVC_ROUTE_HDR_SIZE + pl_len + sig_len; uint8_t* dgram = u_malloc(total_len); if (!dgram) { if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)dgram; hdr->cmd = ETCP_RT_ID_SVC_ROUTE; hdr->dst_node_id = rconn->remote_node_id; hdr->src_node_id = inst->node_id; hdr->seq = seq; hdr->svc_id = rconn->svc_id; { uint8_t f = (rconn->sess_id << ROUTER_SESS_ID_SHIFT) | flag_bits; if (!rconn->start_sent && pl_len > 0 && !flag_bits) { f |= ROUTER_FLAG_START; rconn->start_sent = 1; } if (is_signed) f |= ROUTER_FLAG_SIGNED; hdr->flags = f; } hdr->timestamp = get_current_timestamp(); if (pl_len > 0) memcpy(dgram + SVC_ROUTE_HDR_SIZE, payload, pl_len); struct ETCP_CONN* conn = NULL; struct TOPO_GROUP* group = topo_groups_get_default(inst->topo_groups); if (group) { struct TOPO_NODEQ* nq = topo_node_find_by_id(group, rconn->remote_node_id); if (nq && nq->conn_mgr_type == CONN_TYPE_INDIRECT && nq->conn_mgr_intermediariy_count > 0) { for (uint8_t i = 0; i < nq->conn_mgr_intermediariy_count; i++) { conn = topo_group_find_conn_for_node(group, nq->conn_mgr_intermediaries[i]); if (conn) break; } } } if (!conn) conn = topo_group_find_conn_for_node(group, rconn->remote_node_id); if (!conn) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "no route to %016llx svc_id=%u", (unsigned long long)rconn->remote_node_id, rconn->svc_id); u_free(dgram); if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } if (is_signed) { struct sc_stream_sign_state sign_state; if (sc_stream_sign_init(&conn->crypto_ctx, &sign_state) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "sign_init failed svc_id=%u", rconn->svc_id); u_free(dgram); if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } size_t data_len = SVC_ROUTE_HDR_SIZE + pl_len; if (sc_stream_sign_update(&sign_state, dgram, data_len) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "sign_update failed svc_id=%u", rconn->svc_id); sc_stream_sign_cleanup(&sign_state); u_free(dgram); if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } size_t actual_sig_len = SC_SIGN_SIZE; if (sc_stream_sign_final(&sign_state, dgram + data_len, &actual_sig_len) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "sign_final failed svc_id=%u", rconn->svc_id); u_free(dgram); if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } } struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(dgram); if (!flag_bits) rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; } entry->dgram = dgram; entry->len = total_len; rconn->last_dgram_ts = get_current_timestamp(); DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ETCP_SEND: svc_id=%u seq=%u len=%zu inflight=%d signed=%d → %016llx", rconn->svc_id, seq, pl_len, (int32_t)(rconn->tx_seq - rconn->tx_acked), is_signed, (unsigned long long)rconn->remote_node_id); int ret = etcp_send(conn, entry); if (ret != 0) { if (!flag_bits) rconn->tx_seq--; queue_dgram_free(entry); queue_entry_free(entry); rconn->c_pkts_send_err++; DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_send failed ret=%d svc_id=%u seq=%u", ret, rconn->svc_id, seq); } else { rconn->c_pkts_sent++; if (!flag_bits && pl_len > 0) { struct ROUTER_INFLIGHT* inf = u_malloc(sizeof(struct ROUTER_INFLIGHT)); if (inf) { memset(&inf->ll, 0, sizeof(inf->ll)); inf->ll.size = 4; inf->seq = seq; *(uint32_t*)inf->ll.data = seq; inf->last_sent_tb = get_time_tb(); inf->send_count = 1; inf->payload = u_malloc(pl_len); if (inf->payload) { memcpy(inf->payload, payload, pl_len); inf->payload_len = pl_len; queue_data_put_with_index(rconn->inflight_q, &inf->ll); } else { u_free(inf); } if (!rconn->retrans_timer) router_retrans_schedule(rconn); } } if (!flag_bits) router_track_inflight_state(rconn); } return ret; } // Отправка без conn (RST для неизвестного src) static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t svc_id, uint8_t flag_bits) { struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), dst); if (!conn) return; struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE); if (!hdr) return; hdr->cmd = ETCP_RT_ID_SVC_ROUTE; hdr->dst_node_id = dst; hdr->src_node_id = inst->node_id; hdr->seq = 0; hdr->flags = flag_bits; hdr->svc_id = svc_id; hdr->timestamp = get_current_timestamp(); struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } entry->dgram = (uint8_t*)hdr; entry->len = SVC_ROUTE_HDR_SIZE; DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "flags=0x%02x → %016llx svc_id=%u", flag_bits, (unsigned long long)dst, svc_id); int ret = etcp_send(conn, entry); if (ret != 0) { queue_dgram_free(entry); queue_entry_free(entry); DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_send failed ret=%d", ret); } } static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { struct UTUN_INSTANCE* inst = rconn->inst; etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id]; if (!cb) return; struct ll_entry* e = queue_entry_new(0); if (!e) return; e->dgram = u_malloc(9); if (!e->dgram) { queue_entry_free(e); return; } e->dgram[0] = rconn->svc_id; memcpy(e->dgram + 1, &rconn->remote_node_id, 8); e->len = 9; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx", rconn->svc_id, (unsigned long long)rconn->remote_node_id); cb(NULL, e); } // Закрыть conn: отправить CLOSE удалённой стороне, уведомить локальный сервис, очистить static void router_close_finalize(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; void (*cb)(void*) = rconn->close_callback; void* cb_arg = rconn->close_callback_arg; queue_entry_free(&rconn->ll); if (cb) cb(cb_arg); } static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) { if (!rconn || rconn->closed) return; struct UTUN_INSTANCE* inst = rconn->inst; rconn->closed = 1; router_send_close_to_service(rconn); // Отправляем CLOSE удалённой стороне struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), rconn->remote_node_id); if (conn) router_send_one_flags(rconn, NULL, 0, ROUTER_FLAG_CLOSE, 0); // Таймеры if (rconn->ack_timer) { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; } if (rconn->idle_ack_timer) { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } if (rconn->send_resume_timer) { uasync_cancel_timeout(inst->ua, rconn->send_resume_timer); rconn->send_resume_timer = NULL; } if (rconn->no_route_timer) { uasync_cancel_timeout(inst->ua, rconn->no_route_timer); rconn->no_route_timer = NULL; } if (rconn->recv_q) { struct ll_entry* f; while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } queue_free(rconn->recv_q); rconn->recv_q = NULL; } if (rconn->send_q) { struct ll_entry* f; while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } queue_free(rconn->send_q); rconn->send_q = NULL; } if (rconn->inflight_q) { struct ll_entry* f; while ((f = queue_data_get(rconn->inflight_q)) != NULL) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f; if (inf->payload) u_free(inf->payload); u_free(inf); } queue_free(rconn->inflight_q); rconn->inflight_q = NULL; } if (rconn->retrans_timer) { uasync_cancel_timeout(inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; } if (rconn->incoming_q) { struct ll_entry* f; while ((f = queue_data_get(rconn->incoming_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } queue_resume_callback(rconn->incoming_q); queue_free(rconn->incoming_q); rconn->incoming_q = NULL; } if (inst->router_conns) queue_remove_data(inst->router_conns, &rconn->ll); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: closed remote=%016llx svc_id=%u", (unsigned long long)rconn->remote_node_id, rconn->svc_id); uasync_call_soon(inst->ua, rconn, router_close_finalize); } static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn) { if (rconn->no_route || rconn->closed) return; int sq = queue_entry_count(rconn->send_q); if (sq > 0) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_DRAIN: svc_id=%u send_q=%d inflight=%d", rconn->svc_id, sq, (int32_t)(rconn->tx_seq - rconn->tx_acked)); while (1) { if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) break; struct ll_entry* e = queue_data_get(rconn->send_q); if (!e) { rconn->send_blocked = 0; break; } u_check(e, "drain:after_get", "router"); int s_err = router_send_one(rconn, e->dgram, e->len); u_check(e, "drain:after_send", "router"); queue_dgram_free(e); queue_entry_free(e); if (s_err != 0 && (int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) break; } if (rconn->send_blocked && !rconn->send_resume_timer) rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume"); if (!rconn->send_blocked && rconn->send_resume_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->send_resume_timer); rconn->send_resume_timer = NULL; } } static void router_send_resume_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->send_resume_timer = NULL; router_drain_send_q(rconn); } static void router_no_route_retry_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->no_route_timer = NULL; struct ETCP_CONN* c = topo_group_find_conn_for_node(topo_groups_get_default(rconn->inst->topo_groups), rconn->remote_node_id);; if (c) { DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: route appeared for %016llx svc_id=%u, draining send_q=%d", (unsigned long long)rconn->remote_node_id, rconn->svc_id, queue_entry_count(rconn->send_q)); rconn->no_route = 0; router_drain_send_q(rconn); return; } rconn->no_route_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_NO_ROUTE_RETRY_TB, rconn, router_no_route_retry_cb, "router_no_route"); } // ==================================================================== // Ретрансмиты // ==================================================================== static void router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf) { struct UTUN_INSTANCE* inst = rconn->inst; size_t total_len = SVC_ROUTE_HDR_SIZE + inf->payload_len; uint8_t* dgram = u_malloc(total_len); if (!dgram) { rconn->c_pkts_send_err++; return; } struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)dgram; hdr->cmd = ETCP_RT_ID_SVC_ROUTE; hdr->dst_node_id = rconn->remote_node_id; hdr->src_node_id = inst->node_id; hdr->seq = inf->seq; hdr->svc_id = rconn->svc_id; hdr->flags = (rconn->sess_id << ROUTER_SESS_ID_SHIFT); hdr->timestamp = get_current_timestamp(); struct ETCP_CONN* conn = NULL; struct TOPO_GROUP* group = topo_groups_get_default(inst->topo_groups); if (group) { struct TOPO_NODEQ* nq = topo_node_find_by_id(group, rconn->remote_node_id); if (nq && nq->conn_mgr_type == CONN_TYPE_INDIRECT && nq->conn_mgr_intermediariy_count > 0) { for (uint8_t i = 0; i < nq->conn_mgr_intermediariy_count; i++) { conn = topo_group_find_conn_for_node(group, nq->conn_mgr_intermediaries[i]); if (conn) break; } } } if (!conn) conn = topo_group_find_conn_for_node(group, rconn->remote_node_id); if (!conn) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_retransmit: no route to %016llx svc_id=%u", (unsigned long long)rconn->remote_node_id, rconn->svc_id); u_free(dgram); rconn->c_pkts_send_err++; return; } memcpy(dgram + SVC_ROUTE_HDR_SIZE, inf->payload, inf->payload_len); struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(dgram); rconn->c_pkts_send_err++; return; } entry->dgram = dgram; entry->len = total_len; rconn->last_dgram_ts = get_current_timestamp(); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "RETRANS: svc_id=%u seq=%u attempt=%d → %016llx", rconn->svc_id, inf->seq, inf->send_count + 1, (unsigned long long)rconn->remote_node_id); int ret = etcp_send(conn, entry); if (ret != 0) { queue_dgram_free(entry); queue_entry_free(entry); rconn->c_pkts_send_err++; } else { inf->last_sent_tb = get_time_tb(); inf->send_count++; rconn->c_retrans_done++; rconn->c_pkts_sent++; } } static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn) { if (rconn->retrans_timer) return; uint64_t now = get_time_tb(); if (rconn->last_ack_changed_tb == 0) rconn->last_ack_changed_tb = now; int64_t remaining = (int64_t)ROUTER_RETRANS_TIMEOUT_TB - (int64_t)(now - rconn->last_ack_changed_tb); if (remaining <= 0) remaining = 1; rconn->retrans_timer = uasync_set_timeout(rconn->inst->ua, (uint32_t)remaining, rconn, router_retrans_timer_cb, "router_retrans"); } static void router_retrans_timer_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->retrans_timer = NULL; uint64_t now = get_time_tb(); uint64_t elapsed = now - rconn->last_ack_changed_tb; if (elapsed >= ROUTER_RETRANS_TIMEOUT_TB) { rconn->no_ack_count++; if (rconn->no_ack_count >= ROUTER_NO_ACK_MAX_RETRANS) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no ACK progress for %d retrans cycles, closing svc_id=%u remote=%016llx", rconn->no_ack_count, rconn->svc_id, (unsigned long long)rconn->remote_node_id); router_close_and_notify(rconn); return; } uint32_t s = rconn->tx_acked; while ((int32_t)(s - rconn->tx_acked) < (int32_t)(rconn->tx_seq - rconn->tx_acked)) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)queue_find_data_by_index(rconn->inflight_q, &s); if (!inf) { s++; continue; } router_retransmit_one(rconn, inf); s++; } rconn->last_ack_changed_tb = now; } if (queue_entry_count(rconn->inflight_q) > 0) { int64_t remaining = (int64_t)ROUTER_RETRANS_TIMEOUT_TB - (int64_t)(now - rconn->last_ack_changed_tb); if (remaining <= 0) remaining = 1; rconn->retrans_timer = uasync_set_timeout(rconn->inst->ua, (uint32_t)remaining, rconn, router_retrans_timer_cb, "router_retrans"); } } static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, int force) { if (!force && queue_entry_count(rconn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) { DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: FULL svc_id=%u count=%d — backpressure", rconn->svc_id, queue_entry_count(rconn->send_q)); return -1; } struct ll_entry* qe = queue_entry_new(0); if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "queue_entry_new failed"); return -1; } qe->dgram = u_malloc(pl_len); if (!qe->dgram) { queue_entry_free(qe); return -1; } qe->len = pl_len; if (pl_len > 0) memcpy(qe->dgram, payload, pl_len); DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: queued svc_id=%u inflight=%d send_q=%d", rconn->svc_id, (int32_t)(rconn->tx_seq - rconn->tx_acked), queue_entry_count(rconn->send_q)); queue_data_put(rconn->send_q, qe); u_check(qe, "enqueue_send:after_put", "router"); rconn->send_blocked = 1; if (!rconn->send_resume_timer) rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume"); return 0; } // ==================================================================== // minRTT helpers // ==================================================================== static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn) { return rconn->inflight_limit; } static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) { uint32_t new_limit = rconn->minrtt_probe ? MINRTT_PROBE_MAX_INFLIGHT : rconn->max_inflight; if (new_limit == rconn->inflight_limit) return; uint32_t old = rconn->inflight_limit; rconn->inflight_limit = new_limit; uint32_t inflight = rconn->tx_seq - rconn->tx_acked; if (new_limit < old && inflight >= new_limit) { if (!rconn->send_blocked) rconn->send_blocked = 1; } if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0)) { router_drain_send_q(rconn); } } void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max) { if (!rconn) return; if (new_max < MINRTT_PROBE_MAX_INFLIGHT) new_max = MINRTT_PROBE_MAX_INFLIGHT; rconn->max_inflight = new_max; if (!rconn->minrtt_probe) router_update_inflight_limit(rconn); router_track_inflight_state(rconn); } static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn) { uint32_t inflight = rconn->tx_seq - rconn->tx_acked; uint32_t threshold = rconn->max_inflight / 2; uint8_t is_loaded = (inflight >= threshold); if (is_loaded && !rconn->inflight_was_loaded) { rconn->last_inflight_loaded_tb = get_time_tb(); rconn->inflight_was_loaded = 1; } else if (!is_loaded && rconn->inflight_was_loaded) { rconn->last_inflight_unloaded_tb = get_time_tb(); rconn->inflight_was_loaded = 0; } } static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn) { uint64_t now = get_time_tb(); uint16_t w_idx = rconn->minrtt_window_idx; uint8_t w_cnt = rconn->minrtt_window_count; // Probe check: oldest entry freshness uint8_t prev_probe = rconn->minrtt_probe; if (w_cnt > 0) { uint8_t oldest = (w_idx - w_cnt + MINRTT_WINDOW_SIZE) % MINRTT_WINDOW_SIZE; uint64_t oldest_age = now - rconn->minrtt_window_tb[oldest]; if (w_cnt >= MINRTT_WINDOW_SIZE && oldest_age < MINRTT_PROBE_TIMEOUT_TB) { rconn->minrtt_probe = 0; } else if (oldest_age >= MINRTT_PROBE_TIMEOUT_TB) { rconn->minrtt_probe = 1; } } if (prev_probe != rconn->minrtt_probe) router_update_inflight_limit(rconn); // Update minRTT if channel is unloaded long enough if (!rconn->inflight_was_loaded) { uint64_t unloaded_dur = now - rconn->last_inflight_unloaded_tb; uint64_t min_dur = (uint64_t)rconn->rtt * 2; if (min_dur < MINRTT_UNLOADED_MIN_TB) min_dur = MINRTT_UNLOADED_MIN_TB; if (unloaded_dur >= min_dur) { rconn->minrtt_window[w_idx] = rconn->rtt; rconn->minrtt_window_tb[w_idx] = now; rconn->minrtt_window_idx = (w_idx + 1) % MINRTT_WINDOW_SIZE; if (w_cnt < MINRTT_WINDOW_SIZE) w_cnt++; rconn->minrtt_window_count = w_cnt; uint32_t sum = 0, cnt = 0; for (int i = 0; i < MINRTT_WINDOW_SIZE; i++) { if (rconn->minrtt_window[i] > 0) { sum += rconn->minrtt_window[i]; cnt++; } } rconn->minrtt = cnt > 0 ? (uint16_t)(sum / cnt) : MINRTT_DEFAULT_TB; DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "MINRTT_UPD: svc_id=%u minrtt=%u (rtt=%u cnt=%u probe=%d unloaded=%lu)", rconn->svc_id, rconn->minrtt, rconn->rtt, cnt, rconn->minrtt_probe, (unsigned long)unloaded_dur); } } } // ==================================================================== // ACK timers // ==================================================================== static void router_ack_do_send(struct ETCP_ROUTER_CONN* rconn) { router_send_ack(rconn); } static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { struct UTUN_INSTANCE* inst = rconn->inst; struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), rconn->remote_node_id); if (!conn) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_ack: no route to %016llx svc_id=%u", (unsigned long long)rconn->remote_node_id, rconn->svc_id); rconn->last_ack_sent_tb = get_time_tb(); return; } struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE); if (!hdr) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: u_malloc failed"); return; } hdr->cmd = ETCP_RT_ID_SVC_ROUTE; hdr->dst_node_id = rconn->remote_node_id; hdr->src_node_id = inst->node_id; hdr->seq = rconn->rx_seq; hdr->svc_id = rconn->svc_id; hdr->flags = (rconn->sess_id << ROUTER_SESS_ID_SHIFT); if (rconn->last_recv_updated) { uint64_t now = get_time_tb(); hdr->timestamp = rconn->last_recv_pkt_ts + (uint32_t)(now - rconn->last_recv_pkt_local_tb); rconn->last_recv_updated = 0; } else { hdr->timestamp = 0; } struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } entry->dgram = (uint8_t*)hdr; entry->len = SVC_ROUTE_HDR_SIZE; DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ACK_SEND: svc_id=%u rx_seq=%u → %016llx", rconn->svc_id, rconn->rx_seq, (unsigned long long)rconn->remote_node_id); int ret = etcp_send(conn, entry); if (ret != 0) { queue_dgram_free(entry); queue_entry_free(entry); DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: etcp_send failed ret=%d svc_id=%u rx_seq=%u", ret, rconn->svc_id, rconn->rx_seq); return; } rconn->c_ack_sent++; rconn->last_sent_ack_seq = rconn->rx_seq; rconn->last_ack_sent_tb = get_time_tb(); } static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn) { if (rconn->idle_ack_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } if (rconn->rx_seq == rconn->last_sent_ack_seq) return; uint64_t now = get_time_tb(); uint64_t elapsed = now - rconn->last_ack_sent_tb; if (elapsed >= ROUTER_ACK_INTERVAL_TB) { router_ack_do_send(rconn); return; } if (!rconn->ack_timer) rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_ACK_INTERVAL_TB - (uint32_t)elapsed, rconn, router_ack_timer_cb, "router_ack"); } static void router_ack_timer_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->ack_timer = NULL; if (rconn->rx_seq != rconn->last_sent_ack_seq) router_ack_do_send(rconn); if (rconn->rx_seq != rconn->last_sent_ack_seq) { if (!rconn->ack_timer) rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack"); } else if (!rconn->idle_ack_timer) { rconn->idle_ack_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_ACK_IDLE_TB, rconn, router_idle_ack_timer_cb, "router_idle_ack"); } } static void router_idle_ack_timer_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->idle_ack_timer = NULL; if (rconn->rx_seq != rconn->last_sent_ack_seq) router_ack_do_send(rconn); } // ==================================================================== // Reorder / assembly (аналог etcp_output_try_assembly) // ==================================================================== static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, const uint8_t* payload, size_t payload_len) { struct UTUN_INSTANCE* inst = rconn->inst; etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id]; if (!cb) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: no handler for svc_id=%u", rconn->svc_id); return; } size_t entry_len = 1 + payload_len; struct ll_entry* svc_entry = queue_entry_new(0); if (!svc_entry) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: queue_entry_new failed"); return; } svc_entry->len = entry_len; svc_entry->dgram = u_malloc(entry_len); if (!svc_entry->dgram) { queue_entry_free(svc_entry); return; } svc_entry->dgram[0] = rconn->svc_id; if (payload_len > 0) memcpy(svc_entry->dgram + 1, payload, payload_len); DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: svc_id=%u len=%zu from remote=%016llx", rconn->svc_id, payload_len, (unsigned long long)rconn->remote_node_id); rconn->c_pkts_rcvd++; cb(conn, svc_entry); } static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn) { uint32_t next = rconn->rx_seq; while (1) { struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next); if (!entry) break; queue_remove_data(rconn->recv_q, entry); router_deliver(rconn, conn, entry->dgram, entry->len); rconn->rx_seq = next + 1; next++; queue_dgram_free(entry); queue_entry_free(entry); } } // ==================================================================== // Очередь приёма входящего трафика (между сетью и recv_q) // ==================================================================== static void router_incoming_q_cb(struct ll_queue* q, void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) { queue_resume_callback(q); return; } struct ll_entry* e = queue_data_get(q); if (!e) { queue_resume_callback(q); return; } uint32_t seq = *(uint32_t*)e->data; const uint8_t* pl = e->dgram; size_t pl_len = e->len; if (queue_find_data_by_index(rconn->recv_q, &seq)) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router_incoming_q: dup in recv_q seq=%u, dropping", seq); queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; } struct ll_entry* rq = queue_entry_new(4); if (!rq) { queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; } *(uint32_t*)rq->data = seq; rq->dgram = e->dgram; rq->len = e->len; e->dgram = NULL; queue_entry_free(e); queue_data_put_with_index(rconn->recv_q, rq); struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(rconn->inst->topo_groups), rconn->remote_node_id);; if (seq == rconn->rx_seq) router_try_assembly(rconn, conn); router_schedule_ack(rconn); queue_resume_callback(q); } // ==================================================================== // Извлечённые функции для etcp_router_recv_cb // ==================================================================== static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr, size_t* pl_len) { if (!(hdr->flags & ROUTER_FLAG_SIGNED)) return 0; if (*pl_len < SC_SIGN_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet too short len=%zu from %016llx", *pl_len, (unsigned long long)hdr->src_node_id); free_entry(entry); return -1; } size_t actual_payload = *pl_len - SC_SIGN_SIZE; uint8_t* sig = (uint8_t*)entry->dgram + SVC_ROUTE_HDR_SIZE + actual_payload; struct TOPO_GROUP* group = topo_groups_get_default(inst->topo_groups); struct TOPO_NODEQ* nq = group ? topo_node_find_by_id(group, hdr->src_node_id) : NULL; if (!nq) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet from unknown node %016llx — dropping", (unsigned long long)hdr->src_node_id); free_entry(entry); return -1; } { uint8_t zero[SC_PUBKEY_SIZE] = {0}; if (memcmp(nq->node->ed25519_public_key, zero, SC_PUBKEY_SIZE) == 0) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no Ed25519 pubkey for node %016llx — dropping", (unsigned long long)hdr->src_node_id); free_entry(entry); return -1; } } struct sc_stream_sign_state vfy_state; if (sc_stream_sign_verify_init(&vfy_state, nq->node->ed25519_public_key) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_verify_init failed for node %016llx", (unsigned long long)hdr->src_node_id); free_entry(entry); return -1; } size_t signed_len = SVC_ROUTE_HDR_SIZE + actual_payload; if (sc_stream_sign_update(&vfy_state, entry->dgram, signed_len) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_update failed for node %016llx", (unsigned long long)hdr->src_node_id); sc_stream_sign_cleanup(&vfy_state); free_entry(entry); return -1; } if (sc_stream_sign_verify(&vfy_state, sig, SC_SIGN_SIZE) != SC_OK) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: signature verification FAILED for node %016llx svc_id=%u — dropping", (unsigned long long)hdr->src_node_id, hdr->svc_id); struct ETCP_ROUTER_CONN* sr = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); if (sr) sr->c_sign_fail++; free_entry(entry); return -1; } *pl_len = actual_payload; DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: signature verified for node %016llx svc_id=%u", (unsigned long long)hdr->src_node_id, hdr->svc_id); return 0; } static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts) { (void)src_node_id; if (!rconn) return; if (ack_ts != 0) { uint16_t new_rtt = get_current_timestamp() - ack_ts; int d_rtt = (int)new_rtt - (int)rconn->rtt; if (d_rtt < 0) d_rtt = -d_rtt; rconn->rtt_jitter += ((int32_t)(d_rtt * 65536 - rconn->rtt_jitter)) / 32; rconn->rtt = new_rtt; } if ((int32_t)(seq - rconn->tx_acked) >= 0) { uint32_t old_acked = rconn->tx_acked; rconn->tx_acked = seq; uint32_t clean_seq = old_acked; while ((int32_t)(clean_seq - rconn->tx_acked) < 0) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)queue_find_data_by_index(rconn->inflight_q, &clean_seq); if (inf) { queue_remove_data(rconn->inflight_q, &inf->ll); if (inf->payload) u_free(inf->payload); u_free(inf); } clean_seq++; } rconn->last_ack_changed_tb = get_time_tb(); rconn->no_ack_count = 0; router_track_inflight_state(rconn); router_update_minrtt(rconn); if (queue_entry_count(rconn->inflight_q) == 0 && rconn->retrans_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; } DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_ACKED: svc_id=%u ack=%u inflight=%d send_q=%d inflight_q=%d from %016llx", rconn->svc_id, seq, (int32_t)(rconn->tx_seq - rconn->tx_acked), queue_entry_count(rconn->send_q), queue_entry_count(rconn->inflight_q), (unsigned long long)src_node_id); rconn->c_ack_recv++; } else { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: stale ACK seq=%u tx_acked=%u from %016llx, ignoring", seq, rconn->tx_acked, (unsigned long long)src_node_id); rconn->c_stale_ack++; } rconn->last_dgram_ts = get_current_timestamp(); if (rconn->send_blocked) router_drain_send_q(rconn); } static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr) { struct ETCP_CONN* next = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), hdr->dst_node_id); if (!next) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: no route to %016llx svc_id=%u, dropping", (unsigned long long)hdr->dst_node_id, hdr->svc_id); free_entry(entry); return; } struct TRANSIT_QUEUE* tq = transit_queue_find(next, hdr->src_node_id, hdr->dst_node_id); if (!tq && next->send_input_q && queue_entry_count(next->send_input_q) == 0) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forwarding svc_id=%u → %016llx via %s", hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name); int ret = etcp_send(next, entry); if (ret != 0) free_entry(entry); return; } if (!tq) { tq = transit_queue_get_or_create(next, hdr->src_node_id, hdr->dst_node_id); if (!tq) { free_entry(entry); return; } } int was_empty = (queue_entry_count(tq->q) == 0); queue_data_put(tq->q, entry); DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: queued transit svc_id=%u src=%016llx dst=%016llx q=%d", hdr->svc_id, (unsigned long long)hdr->src_node_id, (unsigned long long)hdr->dst_node_id, queue_entry_count(tq->q)); if (was_empty && next->send_input_q) queue_waiter_wait(next->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); } static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr) { struct ETCP_ROUTER_CONN* rconn = *prconn; uint8_t incoming_sess = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; if (incoming_sess != rconn->peer_sess_id || ((hdr->flags & ROUTER_FLAG_START) && rconn->rx_seq > 0)) { DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: peer restart sess_id=%u→%u svc_id=%u from %016llx", rconn->peer_sess_id, incoming_sess, hdr->svc_id, (unsigned long long)hdr->src_node_id); etcp_router_conn_restart(inst, hdr->src_node_id, hdr->svc_id); *prconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); if (!*prconn) return -1; (*prconn)->peer_sess_id = incoming_sess; rconn = *prconn; } if (hdr->flags & ROUTER_FLAG_RST) { DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: RST from %016llx svc_id=%u", (unsigned long long)hdr->src_node_id, hdr->svc_id); router_close_and_notify(rconn); return -1; } return 0; } static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* pl, size_t pl_len) { (void)conn; rconn->last_dgram_ts = get_current_timestamp(); if (!rconn->peer_sync_done) { rconn->peer_sync_done = 1; rconn->rx_seq = seq; } {// SEQ out of bounds int32_t d = (int32_t)(seq - rconn->rx_seq); if (d > (int32_t)rconn->max_inflight || d < -(int32_t)rconn->max_inflight) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: seq=%u out of bounds, rx_seq=%u (d=%d), dropping", seq, rconn->rx_seq, d); rconn->c_oob_dropped++; return; } } if (((int32_t)(rconn->rx_seq - seq) > 0) || queue_find_data_by_index(rconn->recv_q, &seq)) {// DUP or delta seq<=0 router_schedule_ack(rconn);// сценарий потеряного ack DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: dup seq=%u rx_seq=%u, dropping", seq, rconn->rx_seq); rconn->c_dup_dropped++; return; } struct ll_entry* qe = queue_entry_new(4); if (!qe) return; *(uint32_t*)qe->data = seq; qe->dgram = u_malloc(pl_len); if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); return; } qe->len = pl_len; memcpy(qe->dgram, pl, pl_len); DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router: incoming_q seq=%u rx_seq=%u from %016llx svc_id=%u", seq, rconn->rx_seq, (unsigned long long)rconn->remote_node_id, rconn->svc_id); queue_data_put(rconn->incoming_q, qe); } // ==================================================================== // etcp_router_recv_cb — обработчик ETCP_RT_ID_SVC_ROUTE // ==================================================================== static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { if (!entry) return; struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; if (!inst || entry->len < SVC_ROUTE_HDR_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: invalid packet inst=%p len=%zu min=%u", (void*)inst, entry->len, (unsigned)SVC_ROUTE_HDR_SIZE); free_entry(entry); return; } struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)entry->dgram; size_t pl_len = entry->len - SVC_ROUTE_HDR_SIZE; uint8_t* pl = entry->dgram + SVC_ROUTE_HDR_SIZE; DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ETCP_RECV: svc_id=%u seq=%u len=%zu from %016llx", hdr->svc_id, hdr->seq, pl_len, (unsigned long long)hdr->src_node_id); if (hdr->dst_node_id != inst->node_id) { router_forward_transit(inst, entry, hdr); return; } if (router_verify_signature(inst, entry, hdr, &pl_len) != 0) return; struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); if (pl_len == 0 && (hdr->flags & (ROUTER_FLAG_CLOSE | ROUTER_FLAG_RST))) { if (rconn) router_close_and_notify(rconn); free_entry(entry); return; } if (pl_len == 0) { router_handle_ack(inst, rconn, hdr->seq, hdr->src_node_id, hdr->timestamp); free_entry(entry); return; } if (!rconn) { rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); if (!rconn) { free_entry(entry); return; } rconn->peer_sess_id = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; } if (router_check_peer_restart(inst, &rconn, hdr) != 0) { free_entry(entry); return; } rconn->last_recv_pkt_ts = hdr->timestamp; rconn->last_recv_pkt_local_tb = get_time_tb(); rconn->last_recv_updated = 1; router_handle_data_packet(rconn, conn, hdr->seq, pl, pl_len); free_entry(entry); } // ==================================================================== // Public API // ==================================================================== int etcp_router_init(struct UTUN_INSTANCE* inst) { if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } memset(&inst->router_bindings, 0, sizeof(inst->router_bindings)); inst->router_conns = queue_new(inst->ua, ROUTER_CONN_HASH_SIZE, 0, 9, "router_conns"); if (!inst->router_conns) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "queue_new(router_conns) failed"); return -1; } int ret = etcp_bind(inst, ETCP_ID_SVC_ROUTE, etcp_router_recv_cb); if (ret != 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_bind failed, ret=%d", ret); queue_free(inst->router_conns); inst->router_conns = NULL; return -1; } DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router initialized for node %016llx", (unsigned long long)inst->node_id); return 0; } void etcp_router_destroy(struct UTUN_INSTANCE* inst) { if (!inst) return; etcp_unbind(inst, ETCP_ID_SVC_ROUTE); if (inst->router_conns) { struct ll_entry* entry; while ((entry = queue_data_get(inst->router_conns)) != NULL) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; router_close_and_notify(rconn); } queue_free(inst->router_conns); inst->router_conns = NULL; } memset(&inst->router_bindings, 0, sizeof(inst->router_bindings)); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router destroyed"); } int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback) { if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } if (!callback) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL callback for svc_id=%u", svc_id); return -1; } if (inst->router_bindings.callbacks[svc_id]) DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "overwriting svc_id=%u", svc_id); inst->router_bindings.callbacks[svc_id] = callback; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u → cb=%p", svc_id, (void*)callback); return 0; } int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id) { if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } if (!inst->router_bindings.callbacks[svc_id]) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u not bound", svc_id); return -1; } inst->router_bindings.callbacks[svc_id] = NULL; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u", svc_id); return 0; } int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_entry* entry, int force) { if (!inst || !entry) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL inst=%p entry=%p", (void*)inst, (void*)entry); if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return -1; } if (!entry->dgram || entry->len < 1) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "empty entry: dgram=%p len=%u dst=0x%016llx force=%d", (void*)entry->dgram, entry->len, (unsigned long long)dst_node_id, force); queue_dgram_free(entry); queue_entry_free(entry); return -1; } uint8_t svc_id = entry->dgram[0]; size_t payload_len = entry->len - 1; if (dst_node_id == inst->node_id) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "loopback svc_id=%u len=%zu", svc_id, payload_len); if (inst->router_bindings.callbacks[svc_id]) inst->router_bindings.callbacks[svc_id](NULL, entry); else { queue_dgram_free(entry); queue_entry_free(entry); } return 0; } struct ETCP_CONN* etcp_conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), dst_node_id); if (!etcp_conn) { struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, dst_node_id, svc_id); if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } int eq_ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len, force); queue_dgram_free(entry); queue_entry_free(entry); if (eq_ret == 0) { rconn->no_route = 1; if (!rconn->no_route_timer) rconn->no_route_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_NO_ROUTE_RETRY_TB, rconn, router_no_route_retry_cb, "router_no_route"); return 0; } return -1; } struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, dst_node_id, svc_id); if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { int eq_ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len, force); queue_dgram_free(entry); queue_entry_free(entry); return eq_ret; } int ret = router_send_one(rconn, entry->dgram + 1, payload_len); queue_dgram_free(entry); queue_entry_free(entry); return ret; } int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t node_id) { if (!inst || !topo_groups_get_default(inst->topo_groups)) return -1; struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), node_id); if (!conn || !conn->normalizer || !conn->normalizer->input) return -1; return queue_entry_count(conn->normalizer->input); } void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, struct queue_waiter_handle* h, queue_threshold_callback_fn callback, void* arg) { if (!inst || !topo_groups_get_default(inst->topo_groups) || !h) return; struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), peer_node_id); if (!conn || !conn->normalizer || !conn->normalizer->input) return; queue_waiter_wait(conn->normalizer->input, h, callback, arg); } void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, struct queue_waiter_handle* h) { if (!inst || !topo_groups_get_default(inst->topo_groups) || !h) return; struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_get_default(inst->topo_groups), peer_node_id); if (!conn || !conn->normalizer || !conn->normalizer->input) return; queue_waiter_cancel(conn->normalizer->input, h); } void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) { struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, remote_node_id, svc_id); if (!rconn) return; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — resetting local state", svc_id, (unsigned long long)remote_node_id); etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id]; if (cb) { struct ll_entry* e = queue_entry_new(0); if (e) { e->dgram = u_malloc(9); if (e->dgram) { e->dgram[0] = svc_id; memcpy(e->dgram + 1, &remote_node_id, 8); e->len = 9; cb(NULL, e); } else { queue_entry_free(e); } } } if (rconn->ack_timer) { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; } if (rconn->idle_ack_timer) { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } if (rconn->send_resume_timer) { uasync_cancel_timeout(inst->ua, rconn->send_resume_timer); rconn->send_resume_timer = NULL; } if (rconn->no_route_timer) { uasync_cancel_timeout(inst->ua, rconn->no_route_timer); rconn->no_route_timer = NULL; } if (rconn->retrans_timer) { uasync_cancel_timeout(inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; } struct ll_entry* f; if (rconn->send_q) { while ((f = queue_data_get(rconn->send_q)) != NULL) { u_check(f, "restart:send_q", "router"); queue_dgram_free(f); queue_entry_free(f); } queue_free(rconn->send_q); } if (rconn->recv_q) { while ((f = queue_data_get(rconn->recv_q)) != NULL) { u_check(f, "restart:recv_q", "router"); queue_dgram_free(f); queue_entry_free(f); } queue_free(rconn->recv_q); } if (rconn->inflight_q) { while ((f = queue_data_get(rconn->inflight_q)) != NULL) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f; if (inf->payload) u_free(inf->payload); u_free(inf); } queue_free(rconn->inflight_q); } if (rconn->incoming_q) { while ((f = queue_data_get(rconn->incoming_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } queue_resume_callback(rconn->incoming_q); queue_free(rconn->incoming_q); } rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q"); rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q"); rconn->inflight_q = queue_new(inst->ua, ROUTER_INFLIGHT_HASH_SIZE, 0, 4, "router_inflight_q"); rconn->incoming_q = queue_new(inst->ua, 0, 0, 0, "router_incoming_q"); queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS - 1, 0); queue_set_callback(rconn->incoming_q, router_incoming_q_cb, rconn); queue_set_threshold(rconn->incoming_q, 0, 0); rconn->incoming_data_ready = 0; rconn->tx_seq = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0; rconn->peer_sess_id = 0; rconn->send_blocked = 0; rconn->peer_sync_done = 0; rconn->c_pkts_sent = 0; rconn->c_pkts_send_err = 0; rconn->c_pkts_rcvd = 0; rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_retrans_done = 0; rconn->c_dup_dropped = 0; rconn->c_oob_dropped = 0; rconn->c_stale_ack = 0; rconn->c_sign_fail = 0; rconn->last_ack_changed_tb = 0; rconn->rtt = 0; rconn->rtt_jitter = 0; rconn->max_inflight = ROUTER_MAX_INFLIGHT; rconn->inflight_limit = ROUTER_MAX_INFLIGHT; rconn->minrtt = MINRTT_DEFAULT_TB; memset(rconn->minrtt_window, 0, sizeof(rconn->minrtt_window)); memset(rconn->minrtt_window_tb, 0, sizeof(rconn->minrtt_window_tb)); rconn->minrtt_window_idx = 0; rconn->minrtt_window_count = 0; rconn->minrtt_probe = 0; rconn->last_inflight_loaded_tb = 0; rconn->last_inflight_unloaded_tb = 0; rconn->inflight_was_loaded = 0; rconn->last_recv_pkt_ts = 0; rconn->last_recv_pkt_local_tb = 0; rconn->last_recv_updated = 0; rconn->no_route = 0; rconn->no_ack_count = 0; rconn->closed = 0; } void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h, queue_threshold_callback_fn callback, void* arg) { if (!inst || !h) return; struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id); if (!rconn || !rconn->send_q) return; queue_waiter_wait(rconn->send_q, h, callback, arg); } void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h) { if (!inst || !h) return; struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id); if (!rconn || !rconn->send_q) return; queue_waiter_cancel(rconn->send_q, h); } // ==================================================================== // Seq-connection API // ==================================================================== struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) { if (!inst || !inst->router_conns) return NULL; struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, remote_node_id, svc_id); if (rconn) return rconn; size_t data_size = sizeof(struct ETCP_ROUTER_CONN) - sizeof(struct ll_entry); rconn = (struct ETCP_ROUTER_CONN*)queue_entry_new(data_size); if (!rconn) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_entry_new failed"); return NULL; } rconn->remote_node_id = remote_node_id; rconn->svc_id = svc_id; rconn->last_dgram_ts = get_current_timestamp(); rconn->rtt = 0; rconn->rtt_jitter = 0; rconn->max_inflight = ROUTER_MAX_INFLIGHT; rconn->inflight_limit = ROUTER_MAX_INFLIGHT; rconn->minrtt = MINRTT_DEFAULT_TB; memset(rconn->minrtt_window, 0, sizeof(rconn->minrtt_window)); memset(rconn->minrtt_window_tb, 0, sizeof(rconn->minrtt_window_tb)); rconn->minrtt_window_idx = 0; rconn->minrtt_window_count = 0; rconn->minrtt_probe = 0; rconn->last_inflight_loaded_tb = 0; rconn->last_inflight_unloaded_tb = 0; rconn->inflight_was_loaded = 0; rconn->last_recv_pkt_ts = 0; rconn->last_recv_pkt_local_tb = 0; rconn->last_recv_updated = 0; rconn->tx_seq = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0; rconn->inst = inst; rconn->ack_timer = NULL; rconn->idle_ack_timer = NULL; rconn->send_blocked = 0; rconn->send_resume_timer = NULL; rconn->sess_id = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; rconn->c_pkts_sent = 0; rconn->c_pkts_send_err = 0; rconn->c_pkts_rcvd = 0; rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_dup_dropped = 0; rconn->c_oob_dropped = 0; rconn->c_stale_ack = 0; rconn->c_sign_fail = 0; rconn->c_retrans_done = 0; rconn->retrans_timer = NULL; rconn->last_ack_changed_tb = 0; rconn->no_route = 0; rconn->no_route_timer = NULL; rconn->no_ack_count = 0; rconn->closed = 0; rconn->close_callback = NULL; rconn->close_callback_arg = NULL; rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q"); if (!rconn->recv_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(recv_q) failed"); queue_entry_free(&rconn->ll); return NULL; } rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q"); if (!rconn->send_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(send_q) failed"); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; } rconn->inflight_q = queue_new(inst->ua, ROUTER_INFLIGHT_HASH_SIZE, 0, 4, "router_inflight_q"); if (!rconn->inflight_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(inflight_q) failed"); queue_free(rconn->send_q); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; } rconn->incoming_q = queue_new(inst->ua, 0, 0, 0, "router_incoming_q"); if (!rconn->incoming_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(incoming_q) failed"); queue_free(rconn->inflight_q); queue_free(rconn->send_q); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; } queue_set_callback(rconn->incoming_q, router_incoming_q_cb, rconn); queue_set_threshold(rconn->incoming_q, 0, 0); rconn->incoming_data_ready = 0; queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS - 1, 0); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: new conn remote=%016llx svc_id=%u", (unsigned long long)remote_node_id, svc_id); queue_data_put_with_index(inst->router_conns, &rconn->ll); return rconn; } int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* data, size_t len) { if (!rconn || !data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_send: invalid args rconn=%p data=%p len=%zu", (void*)rconn, (const void*)data, len); return -1; } if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { return router_enqueue_send(rconn, data, len, 0); } return router_send_one(rconn, data, len); } int etcp_router_conn_send_signed(struct ETCP_ROUTER_CONN* rconn, const uint8_t* data, size_t len) { if (!rconn || !data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_send_signed: invalid args rconn=%p data=%p len=%zu", (void*)rconn, (const void*)data, len); return -1; } if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_conn_send_signed: inflight full svc_id=%u", rconn->svc_id); return -1; } return router_send_one_flags(rconn, data, len, 0, 1); } void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn) { router_close_and_notify(rconn); } void etcp_router_conn_close_async(struct ETCP_ROUTER_CONN* rconn, void (*on_close_done)(void* arg), void* close_arg) { if (!rconn) return; rconn->close_callback = on_close_done; rconn->close_callback_arg = close_arg; router_close_and_notify(rconn); } // Reset retransmit state for all router connections to a node // (no callbacks, no close notifications — lightweight, for conn reinit) void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id) { if (!inst || !inst->router_conns) return; struct ll_queue* q = inst->router_conns; for (uint32_t slot = 0; slot < q->hash_size; slot++) { struct ll_entry* entry = q->hash_table[slot]; while (entry) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; if (rconn->remote_node_id == remote_node_id && !rconn->closed) { if (rconn->retrans_timer) { uasync_cancel_timeout(inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; } if (rconn->inflight_q) { struct ll_entry* f; while ((f = queue_data_get(rconn->inflight_q)) != NULL) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f; if (inf->payload) u_free(inf->payload); u_free(inf); } } rconn->no_ack_count = 0; rconn->last_ack_changed_tb = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: paused retrans for remote=%016llx svc_id=%u", (unsigned long long)remote_node_id, rconn->svc_id); } entry = entry->hash_next; } } } void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id) { if (!inst || !inst->router_conns) return; // Iterate via hash chains to avoid inf-loop from put-back in FIFO iteration struct ll_entry* to_close[256]; int count = 0; struct ll_queue* q = inst->router_conns; for (uint32_t slot = 0; slot < q->hash_size && count < 256; slot++) { struct ll_entry* entry = q->hash_table[slot]; while (entry) { struct ll_entry* next = entry->hash_next; struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; if (rconn->remote_node_id == remote_node_id) to_close[count++] = entry; entry = next; } } for (int i = 0; i < count; i++) router_close_and_notify((struct ETCP_ROUTER_CONN*)to_close[i]); }