utun2/src/routing_layer/etcp_router.c

// etcp_router.c — Сервисный слой маршрутизации поверх ETCP
// Упрощённый TCP: восстановление порядка (recv_q), дедупликация, ретрансмиты (inflight_q)
// ACK: периодическая отправка rx_seq (не чаще 100ms), idle-таймер для последнего seq
// Inflight контроль через send_q: при переполнении — очередь + retry-таймер, без дропов
// Подпись/шифрование — в route_crypto.c (encode в начале отправки, decode перед коллбэком)


/*
           /----- ack                                          ack <-- id
           |(pause)                                                     |
in ---> {Q} -> [src, etcp] --> ... --> [dst,etcp] -> {asm_q} -> {buf q} ---> out

*/

#include "etcp_router.h"
#include "route_crypto.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 <string.h>

// ====================================================================
// Внутренние 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* wire, size_t wire_len);
static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flag_bits);
static void router_send_rst(struct ETCP_ROUTER_CONN* rconn);
static struct ETCP_CONN* router_send_conn(struct ETCP_ROUTER_CONN* rconn);
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn);
static void router_send_drain_cb(struct ll_queue* q, void* arg);
static void router_send_watchdog_cb(void* arg);
static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn);
static void router_send_watchdog_disarm(struct ETCP_ROUTER_CONN* rconn);
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 router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn);
static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn);
static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); }
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* wire, size_t wire_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 group_id,
                                                 uint64_t src_node_id, uint64_t dst_node_id) {
    if (!conn->transit_queues) return NULL;
    uint8_t key[24];
    memcpy(key, &group_id, 8);
    memcpy(key + 8, &src_node_id, 8);
    memcpy(key + 16, &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 group_id,
                                                          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, 24, "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, group_id, 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->group_id = group_id;
    tq->src_node_id = src_node_id;
    tq->dst_node_id = dst_node_id;
    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, &group_id, 8);
    memcpy(tq->ll.data + 8, &src_node_id, 8);
    memcpy(tq->ll.data + 16, &dst_node_id, 8);
    queue_data_put_with_index(conn->transit_queues, &tq->ll);
    DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "transit_q: new group=%016llx src=%016llx dst=%016llx",
               (unsigned long long)group_id, (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);
    int dropped = 0;
    struct ll_entry* e;
    while ((e = queue_data_get(tq->q)) != NULL) { free_entry(e); dropped++; }
    if (dropped > 0)
        DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "transit_q: destroyed with %d pending pkts src=%016llx dst=%016llx",
                   dropped, (unsigned long long)tq->src_node_id, (unsigned long long)tq->dst_node_id);
    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) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_drain: etcp_send failed ret=%d src=%016llx dst=%016llx",
                    ret, (unsigned long long)tq->src_node_id, (unsigned long long)tq->dst_node_id);
        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 group_id,
                                                  uint64_t remote_node_id, uint8_t svc_id) {
    if (!inst || !inst->router_conns) return NULL;
    uint8_t key[17];
    memcpy(key, &group_id, 8);
    memcpy(key + 8, &remote_node_id, 8);
    key[16] = svc_id;
    return (struct ETCP_ROUTER_CONN*)queue_find_data_by_index(inst->router_conns, key);
}

// ====================================================================
// Отправка: build+encode в начале send, дренирование через waiter на send_input_q
// ====================================================================

// Найти ETCP_CONN для отправки (учитывая indirect-посредников). NULL = нет маршрута.
static struct ETCP_CONN* router_send_conn(struct ETCP_ROUTER_CONN* rconn) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, rconn->group_id);
    if (group) {
        struct TOPO_GROUP_NODE* 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++) {
                struct ETCP_CONN* c = topo_group_find_conn_for_node(group, nq->conn_mgr_intermediaries[i]);
                if (c) return c;
            }
        }
    }
    return topo_group_find_conn_for_node(group, rconn->remote_node_id);
}

// Собрать и закодировать (подпись/шифрование) пакет в начале отправки.
// Возвращает финальный wire-пакет [SVC_ROUTE_HDR][payload][sig?] (u_malloc).
static int router_build_packet(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, int mode,
                               uint8_t** out, size_t* out_len) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    uint8_t* base = u_malloc(SVC_ROUTE_HDR_SIZE + pl_len);
    if (!base) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_build_packet: alloc fail"); rconn->c_pkts_send_err++; return -1; }
    struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)base;
    hdr->cmd         = ETCP_RT_ID_SVC_ROUTE;
    hdr->group_id    = rconn->group_id;
    hdr->dst_node_id = rconn->remote_node_id;
    hdr->src_node_id = inst->node_id;
    hdr->seq         = rconn->tx_seq++;
    hdr->svc_id      = rconn->svc_id;
    {
        uint8_t f = (rconn->sess_id << ROUTER_SESS_ID_SHIFT);
        if (!rconn->start_sent && hdr->seq == 0 && pl_len > 0) f |= ROUTER_FLAG_START;
        hdr->flags = f;
    }
    hdr->timestamp = get_current_timestamp();
    if (pl_len > 0) memcpy(base + SVC_ROUTE_HDR_SIZE, payload, pl_len);

    if (route_crypto_encode(inst, rconn->remote_node_id, base, SVC_ROUTE_HDR_SIZE + pl_len, mode, out, out_len) != 0) {
        u_free(base); rconn->tx_seq--; rconn->c_pkts_send_err++; return -1;
    }
    u_free(base);
    return 0;
}

// Отправить контрольный (header-only) пакет: CLOSE и т.п. Без seq/подписи/шифрования.
static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flag_bits) {
    struct ETCP_CONN* conn = router_send_conn(rconn);
    if (!conn) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_ctrl: no route to %016llx svc_id=%u flag=%02x",
                   (unsigned long long)rconn->remote_node_id, rconn->svc_id, flag_bits);
        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->group_id    = rconn->group_id;
    hdr->dst_node_id = rconn->remote_node_id;
    hdr->src_node_id = rconn->inst->node_id;
    hdr->seq         = 0;
    hdr->svc_id      = rconn->svc_id;
    hdr->flags       = (rconn->sess_id << ROUTER_SESS_ID_SHIFT) | flag_bits;
    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;
    int ret = etcp_send(conn, entry);
    if (ret != 0) {
        queue_dgram_free(entry); queue_entry_free(entry);
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ctrl: etcp_send failed ret=%d svc_id=%u flag=%02x", ret, rconn->svc_id, flag_bits);
    }
}

// Отправка RST удалённой стороне — сброс чужой «старой» сессии.
// Идёт через recv_conn: в асимметричной топологии route-lookup до remote вернёт NULL.
static void router_send_rst(struct ETCP_ROUTER_CONN* rconn) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    struct ETCP_CONN* conn = rconn->recv_conn;
    if (!conn) conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, rconn->group_id), rconn->remote_node_id);
    if (!conn) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_rst: no route to %016llx svc_id=%u",
                   (unsigned long long)rconn->remote_node_id, rconn->svc_id);
        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->group_id    = rconn->group_id;
    hdr->dst_node_id = rconn->remote_node_id;
    hdr->src_node_id = inst->node_id;
    hdr->seq         = 0;
    hdr->flags       = ROUTER_FLAG_RST | (rconn->sess_id << ROUTER_SESS_ID_SHIFT);
    hdr->svc_id      = rconn->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_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_rst: → %016llx svc_id=%u",
               (unsigned long long)rconn->remote_node_id, rconn->svc_id);
    int ret = etcp_send(conn, entry);
    if (ret != 0) {
        queue_dgram_free(entry); queue_entry_free(entry);
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_rst: etcp_send failed ret=%d", ret);
    }
}

static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) {
    static struct ETCP_CONN c;
    memset(&c, 0, sizeof(c));
    c.instance = inst; c.peer_node_id = inst->node_id;
    return &c;
}

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(ROUTER_SVC_HDR_SIZE);
    if (!e->dgram) { queue_entry_free(e); return; }
    e->dgram[0] = rconn->svc_id;
    memcpy(e->dgram + ROUTER_SVC_SRC_OFF, &rconn->remote_node_id, 8);
    memcpy(e->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8);
    e->dgram[ROUTER_SVC_FLAGS_OFF] = 0;
    e->len = ROUTER_SVC_HDR_SIZE;
    DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx",
               rconn->svc_id, (unsigned long long)rconn->remote_node_id);
    cb(loopback_conn(rconn->inst), 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);
}

// Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart).
static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    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->watchdog_timer) { uasync_cancel_timeout(inst->ua, rconn->watchdog_timer); rconn->watchdog_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 ETCP_CONN* c = router_send_conn(rconn);
        if (c && c->send_input_q) queue_waiter_cancel(c->send_input_q, &rconn->send_waiter);
        memset(&rconn->send_waiter, 0, sizeof(rconn->send_waiter));
    }
    struct ll_entry* f;
    if (rconn->send_q) {
        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->recv_q) {
        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->inflight_q) {
        while ((f = queue_data_get(rconn->inflight_q)) != NULL) {
            struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f;
            if (inf->dgram) u_free(inf->dgram);
            u_free(inf);
        }
        queue_free(rconn->inflight_q); rconn->inflight_q = NULL;
    }
    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->incoming_q = NULL;
    }
}

// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/close_cb/last_dgram_ts).
static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) {
    router_conn_free_queues(rconn);

    rconn->send_q     = queue_new(rconn->inst->ua, 0, 0, 0, "router_send_q");
    rconn->recv_q     = queue_new(rconn->inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q");
    rconn->inflight_q = queue_new(rconn->inst->ua, ROUTER_INFLIGHT_HASH_SIZE, 0, 4, "router_inflight_q");
    rconn->incoming_q = queue_new(rconn->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->recv_conn = NULL;

    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->peer_sess_id = 0;
    rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0;
    rconn->no_route = 0; rconn->no_route_timer = NULL; rconn->no_ack_count = 0; rconn->closed = 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;
}

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 удалённой стороне + снимаем waiter отправки
    if (router_send_conn(rconn)) router_send_ctrl(rconn, ROUTER_FLAG_CLOSE);
    router_conn_free_queues(rconn);
    if (inst->router_conns)
        queue_remove_data(inst->router_conns, &rconn->ll);
    DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: closed group=%016llx remote=%016llx svc_id=%u",
               (unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id);
    uasync_call_soon(inst->ua, rconn, router_close_finalize);
}

// Пометить отсутствие маршрута и взвести таймер повторных проверок (идемпотентно).
static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) {
    if (rconn->no_route) return;
    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");
}

// Попытаться продвинуть отправку: зарегистрировать waiter на send_input_q, если есть что слать.
// Инвариант: send_q непуст ∧ !closed ∧ !no_route ∧ inflight свободен ⇒ waiter зарегистрирован.
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) {
    if (rconn->closed || rconn->no_route) return;
    if (queue_entry_count(rconn->send_q) == 0) return;
    if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) {
        rconn->send_blocked = 1;
        return;
    }
    if (rconn->send_waiter.internal) return;
    struct ETCP_CONN* conn = router_send_conn(rconn);
    if (!conn) { router_set_no_route(rconn); return; }
    queue_waiter_wait(conn->send_input_q, &rconn->send_waiter, router_send_drain_cb, rconn);
}

// Waiter-коллбэк: send_input_q опустел → шлём один пакет (уже закодированный), при необходимости
// снова встаём в хвост (round-robin). Успешно отправленный пакет кладём в inflight_q для ретрансмита.
static void router_send_drain_cb(struct ll_queue* q, void* arg) {
    (void)q;
    struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
    if (rconn->closed || rconn->no_route) return;
    if (queue_entry_count(rconn->send_q) == 0) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; }
    if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) {
        rconn->send_blocked = 1;
        return;
    }
    struct ll_entry* e = queue_data_get(rconn->send_q);
    if (!e) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; }
    uint32_t seq = ((struct SVC_ROUTE_HDR*)e->dgram)->seq;

    struct ETCP_CONN* conn = router_send_conn(rconn);
    if (!conn) { free_entry(e); router_set_no_route(rconn); return; }

    // копия финального пакета для inflight (e->dgram уйдёт в etcp_send)
    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->dgram = u_malloc(e->len);
        if (inf->dgram) { memcpy(inf->dgram, e->dgram, e->len); inf->dgram_len = e->len; }
        else { u_free(inf); inf = NULL; }
    }

    DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_DRAIN: svc_id=%u send_q=%d inflight=%d sq_in=%d seq=%u",
                rconn->svc_id, queue_entry_count(rconn->send_q),
                queue_entry_count(rconn->inflight_q), rconn->send_q->count, seq);
    int ret = etcp_send(conn, e);
    if (ret != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_send failed ret=%d svc_id=%u seq=%u", ret, rconn->svc_id, seq);
        free_entry(e); rconn->c_pkts_send_err++;
        if (inf) { if (inf->dgram) u_free(inf->dgram); u_free(inf); }
    } else {
        rconn->c_pkts_sent++;
        rconn->last_dgram_ts = get_current_timestamp();
        if (inf) {
            queue_data_put_with_index(rconn->inflight_q, &inf->ll);
            if (!rconn->retrans_timer) router_retrans_schedule(rconn);
        }
        router_track_inflight_state(rconn);
    }

    if (queue_entry_count(rconn->send_q) == 0) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; }
    if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) {
        rconn->send_blocked = 1;
        return;
    }
    conn = router_send_conn(rconn);
    if (!conn) { router_set_no_route(rconn); return; }
    queue_waiter_wait(conn->send_input_q, &rconn->send_waiter, router_send_drain_cb, rconn);
}

// DEBUG-watchdog: медленная (500мс) проверка инварианта drain. Если send_q непуст,
// канал свободен, но waiter не зарегистрирован — инвариант нарушен: логируем и форсируем kick.
static void router_send_watchdog_cb(void* arg) {
    struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
    rconn->watchdog_timer = NULL;
    if (rconn->closed) return;
    if (queue_entry_count(rconn->send_q) == 0) return;
    if (!rconn->no_route
        && queue_entry_count(rconn->inflight_q) < (int)router_effective_max_inflight(rconn)
        && !rconn->send_waiter.internal && !rconn->send_waiter.call_soon_id) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router watchdog: send_q stalled svc_id=%u send_q=%d inflight=%d blocked=%d — force kick",
                    rconn->svc_id, queue_entry_count(rconn->send_q),
                    queue_entry_count(rconn->inflight_q), rconn->send_blocked);
        router_send_kick(rconn);
    }
    if (queue_entry_count(rconn->send_q) > 0)
        rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog");
}

// Взвести watchdog пока send_q непуст (страховка от пропущенного re-arm).
static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn) {
    if (rconn->watchdog_timer || queue_entry_count(rconn->send_q) == 0) return;
    rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua,
        ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog");
}

// Снять watchdog когда send_q опустел.
static void router_send_watchdog_disarm(struct ETCP_ROUTER_CONN* rconn) {
    if (rconn->watchdog_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->watchdog_timer); rconn->watchdog_timer = NULL; }
}

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;
    if (router_send_conn(rconn)) {
        DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: route appeared for %016llx svc_id=%u, resuming send_q=%d",
                   (unsigned long long)rconn->remote_node_id, rconn->svc_id, queue_entry_count(rconn->send_q));
        rconn->no_route = 0;
        router_send_kick(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 ETCP_CONN* conn = router_send_conn(rconn);
    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);
        rconn->c_pkts_send_err++; return;
    }
    // Финальный пакет уже закодирован (подпись/шифрование) — шлём как есть.
    struct ll_entry* entry = queue_entry_new(0);
    if (!entry) { rconn->c_pkts_send_err++; return; }
    entry->dgram = u_malloc(inf->dgram_len);
    if (!entry->dgram) { queue_entry_free(entry); rconn->c_pkts_send_err++; return; }
    memcpy(entry->dgram, inf->dgram, inf->dgram_len);
    entry->len = inf->dgram_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");
    }
}

// Собрать+закодировать пакет в начале отправки и положить в send_q.
static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, int force, int mode) {
    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;
    }
    uint8_t* dgram = NULL; size_t dgram_len = 0;
    if (router_build_packet(rconn, payload, pl_len, mode, &dgram, &dgram_len) != 0) return -1;

    struct ll_entry* qe = queue_entry_new(0);
    if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "queue_entry_new failed"); u_free(dgram); rconn->tx_seq--; rconn->c_pkts_send_err++; return -1; }
    qe->dgram = dgram;
    qe->len   = dgram_len;
    DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: queued svc_id=%u inflight=%d send_q=%d mode=%02x",
                rconn->svc_id, queue_entry_count(rconn->inflight_q), queue_entry_count(rconn->send_q), mode);
    queue_data_put(rconn->send_q, qe);
    router_send_kick(rconn);
    router_send_watchdog_arm(rconn);
    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 = (uint32_t)queue_entry_count(rconn->inflight_q);

    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)) {
        rconn->send_blocked = 0;
        router_send_kick(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 = (uint32_t)queue_entry_count(rconn->inflight_q);
    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 = rconn->recv_conn;
    if (!conn) conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, rconn->group_id), 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->group_id    = rconn->group_id;
    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)
// ====================================================================

// Доставка сервисной кодограммы в едином формате [svc_id][src][dst][rx_flags][payload].
// Для loopback (dst == self) src и dst оба = self, rx_flags = 0.
static void router_deliver_loopback(struct UTUN_INSTANCE* inst, uint8_t svc_id, struct ll_entry* entry) {
    etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id];
    if (!cb) { queue_dgram_free(entry); queue_entry_free(entry); return; }
    size_t payload_len = entry->len - 1;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { queue_dgram_free(entry); queue_entry_free(entry); return; }
    e->dgram = u_malloc(ROUTER_SVC_HDR_SIZE + payload_len);
    if (!e->dgram) { queue_entry_free(e); queue_dgram_free(entry); queue_entry_free(entry); return; }
    e->dgram[0] = svc_id;
    memcpy(e->dgram + ROUTER_SVC_SRC_OFF, &inst->node_id, 8);
    memcpy(e->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8);
    e->dgram[ROUTER_SVC_FLAGS_OFF] = 0;
    if (payload_len > 0) memcpy(e->dgram + ROUTER_SVC_PAYLOAD_OFF, entry->dgram + 1, payload_len);
    e->len = ROUTER_SVC_HDR_SIZE + payload_len;
    queue_dgram_free(entry); queue_entry_free(entry);
    cb(loopback_conn(inst), e);
}

// Доставка: декодирует (подпись/шифрование) перед вызовом сервисного коллбэка.
// wire = [SVC_ROUTE_HDR][payload][sig?] — финальный пакет, лежавший в recv_q.
static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn,
                           const uint8_t* wire, size_t wire_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;
    }
    const uint8_t* payload; size_t payload_len; uint8_t rx_flags = 0; uint8_t* decoded = NULL;
    struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)wire;
    if (hdr->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) {
        size_t decoded_len = 0;
        if (route_crypto_decode(inst, wire, wire_len, &decoded, &decoded_len, &rx_flags) != 0) {
            DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: decode failed svc_id=%u from %016llx — dropping",
                       rconn->svc_id, (unsigned long long)rconn->remote_node_id);
            rconn->c_sign_fail++;
            return;
        }
        payload     = decoded + SVC_ROUTE_HDR_SIZE;
        payload_len = decoded_len - SVC_ROUTE_HDR_SIZE;
    } else {
        payload     = wire + SVC_ROUTE_HDR_SIZE;
        payload_len = wire_len - SVC_ROUTE_HDR_SIZE;
    }

    size_t entry_len = ROUTER_SVC_HDR_SIZE + 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"); if (decoded) u_free(decoded); return; }
    svc_entry->len   = entry_len;
    svc_entry->dgram = u_malloc(entry_len);
    if (!svc_entry->dgram) { queue_entry_free(svc_entry); if (decoded) u_free(decoded); return; }
    svc_entry->dgram[0] = rconn->svc_id;
    memcpy(svc_entry->dgram + ROUTER_SVC_SRC_OFF, &rconn->remote_node_id, 8);
    memcpy(svc_entry->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8);
    svc_entry->dgram[ROUTER_SVC_FLAGS_OFF] = rx_flags;
    if (payload_len > 0) memcpy(svc_entry->dgram + ROUTER_SVC_PAYLOAD_OFF, payload, payload_len);
    if (decoded) u_free(decoded);

    DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: svc_id=%u len=%zu flags=%02x src=%016llx dst=%016llx",
                rconn->svc_id, payload_len, rx_flags, (unsigned long long)rconn->remote_node_id,
                (unsigned long long)inst->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* wire = e->dgram;
    size_t wire_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);

    if (seq == rconn->rx_seq) router_try_assembly(rconn, rconn->recv_conn);
    router_schedule_ack(rconn);
    queue_resume_callback(q);
}

// ====================================================================
// Функции приёма etcp_router_recv_cb
// ====================================================================

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;
        if (rconn->tx_acked > 0) rconn->start_sent = 1;// первый ACK подтвердил seq 0 — сессия больше не «новая»
        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->dgram) u_free(inf->dgram);
                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, queue_entry_count(rconn->inflight_q),
                    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) { rconn->send_blocked = 0; router_send_kick(rconn); }
}

static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry,
                                    struct SVC_ROUTE_HDR* hdr) {
    struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, hdr->group_id);
    struct ETCP_CONN* next = topo_group_find_conn_for_node(group, hdr->dst_node_id);
    if (!next) {
        int node_count = group && group->nodes ? queue_entry_count(group->nodes) : -1;
        DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: no route to %016llx svc_id=%u, dropping (group=%016llx nodes=%d)",
                   (unsigned long long)hdr->dst_node_id, hdr->svc_id,
                   (unsigned long long)hdr->group_id, node_count);
        if (group && group->nodes && node_count > 0 && node_count < 10) {
            char list[512]; int pos = 0;
            struct ll_entry* e = group->nodes->head;
            while (e) {
                struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
                pos += snprintf(list + pos, sizeof(list) - pos, "%s%016llx(up=%02x)",
                                pos ? "," : "", (unsigned long long)nq->node_id, nq->conn_up);
                e = e->next;
            }
            DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: group nodes: %s", list);
        }
        free_entry(entry); return;
    }
    struct TRANSIT_QUEUE* tq = transit_queue_find(next, hdr->group_id, 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) {
            DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forward etcp_send failed ret=%d svc_id=%u → %016llx",
                        ret, hdr->svc_id, (unsigned long long)hdr->dst_node_id);
            free_entry(entry);
        }
        return;
    }

    if (!tq) {
        tq = transit_queue_get_or_create(next, hdr->group_id, 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->group_id, hdr->src_node_id, hdr->svc_id);
        *prconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
        if (!*prconn) return -1;
        (*prconn)->peer_sess_id = incoming_sess;
        rconn = *prconn;
    }
    return 0;
}

static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn,
                                       uint32_t seq, const uint8_t* wire, size_t wire_len) {
    rconn->recv_conn = 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
        // Сценарий потерянного ACK: пир ретранслит, значит не получил наш ACK — дослать его.
        // router_schedule_ack здесь не годится: после первого ACK rx_seq==last_sent_ack_seq и он выходит рано.
        uint64_t now = get_time_tb();
        if (now - rconn->last_ack_sent_tb >= ROUTER_ACK_INTERVAL_TB) {
            DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: dup seq=%u rx_seq=%u, resending ACK", seq, rconn->rx_seq);
            router_send_ack(rconn);
        }
        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(wire_len);
    if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); return; }
    qe->len = wire_len; memcpy(qe->dgram, wire, wire_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;

    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; }

    struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);

    if (pl_len == 0 && (hdr->flags & ROUTER_FLAG_RST)) {
        // Пир отклонил нашу «старую» сессию — сброс локального состояния (пере-шлём START).
        if (rconn) etcp_router_conn_restart(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
        free_entry(entry); return;
    }
    if (pl_len == 0 && (hdr->flags & ROUTER_FLAG_CLOSE)) {
        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->group_id, 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; }

    // Сервер после рестарта «ждёт START»: пришёл не-START пакет от старой сессии → RST.
    if (!rconn->peer_sync_done && !(hdr->flags & ROUTER_FLAG_START)) {
        router_send_rst(rconn);
        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, entry->dgram, entry->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, 17, "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_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "unbind svc_id=%u", svc_id);
        inst->router_bindings.callbacks[svc_id] = NULL;
        return 0;
    }
    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 group_id, uint64_t dst_node_id, struct ll_entry* entry, int force, int mode) {
    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);
        router_deliver_loopback(inst, svc_id, entry);
        return 0;
    }

    struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, group_id, dst_node_id, svc_id);
    if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }

    int ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len, force, mode);
    queue_dgram_free(entry); queue_entry_free(entry);
    return ret;
}

int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id) {
    if (!inst || !topo_groups_find(inst->topo_groups, group_id)) return -1;
    struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), 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 group_id, uint64_t peer_node_id,
                                  struct queue_waiter_handle* h,
                                  queue_threshold_callback_fn callback, void* arg) {
    if (!inst || !topo_groups_find(inst->topo_groups, group_id) || !h) return;
    struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), 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 group_id, uint64_t peer_node_id,
                                struct queue_waiter_handle* h) {
    if (!inst || !topo_groups_find(inst->topo_groups, group_id) || !h) return;
    struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), 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 group_id, uint64_t remote_node_id, uint8_t svc_id) {
    struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, 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(ROUTER_SVC_HDR_SIZE);
            if (e->dgram) {
                e->dgram[0] = svc_id;
                memcpy(e->dgram + ROUTER_SVC_SRC_OFF, &remote_node_id, 8);
                memcpy(e->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8);
                e->dgram[ROUTER_SVC_FLAGS_OFF] = 0;
                e->len = ROUTER_SVC_HDR_SIZE;
                cb(loopback_conn(inst), e);
            } else { queue_entry_free(e); }
        }
    }

    router_conn_reset(rconn);
}

void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, 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, group_id, 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 group_id, 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, group_id, node_id, svc_id);
    if (!rconn || !rconn->send_q) return;
    queue_waiter_cancel(rconn->send_q, h);
}

int etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id) {
    if (!inst) return 0;
    struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, node_id, svc_id);
    if (!rconn || !rconn->send_q) return 0;
    return rconn->send_q->count <= rconn->send_q->threshold_max_packets
        && (rconn->send_q->threshold_max_bytes == 0 || rconn->send_q->total_bytes <= rconn->send_q->threshold_max_bytes);
}

// ====================================================================
// Seq-connection API
// ====================================================================

struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst,
                                               uint64_t group_id,
                                               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, group_id, 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->group_id       = group_id;
    rconn->remote_node_id = remote_node_id;
    rconn->svc_id         = svc_id;
    rconn->inst           = inst;
    memcpy(rconn->ll.data, &group_id, 8);
    memcpy(rconn->ll.data + 8, &remote_node_id, 8);
    rconn->ll.data[16] = svc_id;
    rconn->last_dgram_ts  = get_current_timestamp();
    router_conn_reset(rconn);
    if (!rconn->send_q || !rconn->recv_q || !rconn->inflight_q || !rconn->incoming_q) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new failed");
        router_conn_free_queues(rconn);
        queue_entry_free(&rconn->ll);
        return NULL;
    }
    DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: new conn group=%016llx remote=%016llx svc_id=%u",
               (unsigned long long)group_id, (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, int mode) {
    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;
    }
    return router_enqueue_send(rconn, data, len, 0, mode);
}

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->watchdog_timer) { uasync_cancel_timeout(inst->ua, rconn->watchdog_timer); rconn->watchdog_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->dgram) u_free(inf->dgram);
                        u_free(inf);
                    }
                }
                {
                    struct ETCP_CONN* c = router_send_conn(rconn);
                    if (c && c->send_input_q) queue_waiter_cancel(c->send_input_q, &rconn->send_waiter);
                    memset(&rconn->send_waiter, 0, sizeof(rconn->send_waiter));
                }
                rconn->send_blocked = 0;
                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 group_id, 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->group_id == group_id && 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]);
}