From 3bd9b964fa0a2e0bd642aacd1296c001e47d9474 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 15:07:41 +0200 Subject: [PATCH] Remove obsolete router state and clarify receive diagnostics --- src/routing_layer/etcp_router.c | 38 +++++++++++++++++---------------- src/routing_layer/etcp_router.h | 2 -- tests/test_etcp_router_unit.c | 3 +-- 3 files changed, 21 insertions(+), 22 deletions(-) diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index faeca5d4..27e35cdc 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -5,7 +5,7 @@ // Подпись/шифрование — в route_crypto.c (encode при отправке, decode до изменения состояния приёма) // // Поток сервисной кодограммы: -// in -> {send_q} -> [src: etcp_send] -> ... -> [dst: recv_q] -> {incoming_q} -> out +// in -> {send_q} -> {inflight_q / etcp_send} -> ... -> {incoming_q} -> {recv_q} -> сервис // ack возвращается по обратному пути; (pause) — backpressure при заполнении очередей. @@ -431,7 +431,6 @@ static void router_confirm_peer(struct ETCP_ROUTER_CONN* rconn, const struct SVC if (previous && router_conn_reset(rconn) != 0) { router_close_and_notify(rconn); return; } if (rconn->handshake_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } rconn->peer_reset_id = peer_id; - rconn->peer_sync_done = 1; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router session: group=%016llx peer=%016llx svc=%u local=%016llx remote=%016llx previous=%016llx", (unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, @@ -446,9 +445,12 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { 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; + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router close notification allocation failed svc=%u", rconn->svc_id); return; } e->dgram = u_malloc(ROUTER_SVC_HDR_SIZE); - if (!e->dgram) { queue_entry_free(e); return; } + if (!e->dgram) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router close notification payload allocation failed svc=%u", rconn->svc_id); + 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); @@ -522,7 +524,6 @@ static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { 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->receive_blocked = 0; rconn->peer_receive_blocked = 0; rconn->recv_conn = NULL; @@ -540,7 +541,7 @@ static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { rconn->tx_seq = 0; rconn->tx_sent = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; rconn->retrans_seq = 0; rconn->retrans_end = 0; rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0; - rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; + rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_reset_id = 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; @@ -551,8 +552,7 @@ static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { return 0; } -// Закрыть conn: уведомить локальный сервис, отправить CLOSE удалённой стороне, очистить -// очереди/таймеры, удалить из реестра router_conns и отложить освобождение ll через call_soon. +// Снять канал с реестра, уведомить сервис/пира, очистить очереди/таймеры; само ll освободить через call_soon. static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) { if (!rconn || rconn->closed) return; struct UTUN_INSTANCE* inst = rconn->inst; @@ -799,7 +799,7 @@ static void router_retrans_timer_cb(void* arg) { } } -// Собрать+закодировать пакет в начале отправки и положить в send_q. +// Скопировать plaintext и mode в send_q, выделить seq; кодирование выполняется при отправке после handshake. static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t len, int force, int mode) { uint8_t required = rconn->inst->router_bindings.required_crypto[rconn->svc_id]; if ((mode & required) != required) { @@ -1061,8 +1061,7 @@ static int router_deliver_loopback(struct UTUN_INSTANCE* inst, uint64_t group_id return 0; } -// Доставка: декодирует (подпись/шифрование) перед вызовом сервисного коллбэка. -// wire = [SVC_ROUTE_HDR][payload][sig?] — финальный пакет, лежавший в recv_q. +// Доставить уже проверенный plaintext из recv_q; криптографические проверки выполнены до постановки в очередь. static int 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; @@ -1144,8 +1143,6 @@ static void router_incoming_q_cb(struct ll_queue* q, void* arg) { 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 ((int32_t)(seq - rconn->rx_seq) < 0 || queue_find_data_by_index(rconn->recv_q, &seq)) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_incoming_q: dup in recv_q seq=%u, dropping", seq); @@ -1153,7 +1150,10 @@ static void router_incoming_q_cb(struct ll_queue* q, void* arg) { } struct ll_entry* rq = queue_entry_new(4); - if (!rq) { queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; } + if (!rq) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router reorder entry allocation failed svc=%u seq=%u", rconn->svc_id, seq); + 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); @@ -1276,8 +1276,7 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* transit_queue_wait(tq); } -// Принять data-пакет: зафиксировать recv_conn, синхронизировать rx_seq при первом пакете сеанса, -// отсеять out-of-bounds/дубликаты и положить пакет в incoming_q. +// Принять данные согласованной сессии: зафиксировать recv_conn, проверить окно/дубликаты и положить в incoming_q. 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; @@ -1311,10 +1310,13 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC return; } struct ll_entry* qe = queue_entry_new(4); - if (!qe) return; + if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router incoming entry allocation failed svc=%u seq=%u", rconn->svc_id, seq); return; } *(uint32_t*)qe->data = seq; qe->dgram = u_malloc(wire_len); - if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); return; } + if (!qe->dgram) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router incoming payload allocation failed svc=%u seq=%u bytes=%zu", rconn->svc_id, seq, wire_len); + 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", diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 80be1a3e..a6440a2e 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -156,7 +156,6 @@ struct ETCP_ROUTER_CONN { uint64_t challenge_sent_tb; // время отправки запроса, 0.1 ms void* handshake_timer; // таймер повторов согласования uint8_t start_sent; // получен первый ACK данных - uint8_t peer_sync_done; // 1 = завершено подтверждение peer ID // Ретрансмиты: inflight очередь struct ll_queue* inflight_q; // хеш по seq (4 байта), непрерывный блок без SACK @@ -165,7 +164,6 @@ struct ETCP_ROUTER_CONN { // Очередь приёма входящего трафика (между сетью и recv_q) struct ll_queue* incoming_q; // FIFO очередь входящих пакетов - uint8_t incoming_data_ready; // 1 = данные в incoming_q и callback был вызван uint8_t receive_blocked; // сервис временно не принимает следующую кодограмму uint8_t peer_receive_blocked; // последний ACK пира объявил паузу приёма void* receive_pump; // отложенное продолжение приёма diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index e00420e9..6b4934f8 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -526,7 +526,7 @@ static int test_out_of_bounds(void) { rx_reset(0xEE); uint8_t data[] = { TEST_SVC_ID, 0xEE, 0 }; - /* first packet syncs rx_seq (peer_sync_done), then OOB check works */ + /* После handshake первый DATA начинается с seq=0, затем проверяется граница окна. */ inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START); uint32_t bad_seq = ROUTER_MAX_INFLIGHT + 2; inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, bad_seq, data + 1, 2, 0, 0); @@ -548,7 +548,6 @@ static int test_circular_wrap(void) { if (!c) FAIL("conn_get failed"); c->rx_seq = 0xFFFFFFFE; - c->peer_sync_done = 1;// сессия уже установлена (seq вручную на границе wrap) rx_reset(0x11); rx.expected_seq = 0xFFFFFFFE; uint8_t data[] = { TEST_SVC_ID, 0x11, 0 };