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

Remove obsolete router state and clarify receive diagnostics

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 15:07:41 +02:00
1 parent a258b77240
commit 3bd9b964fa
3 files changed
+21 -22
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/etcp_router.c
+20 -18
View File
@@ -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",
src/routing_layer/etcp_router.h
-2
View File
@@ -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; // отложенное продолжение приёма
tests/test_etcp_router_unit.c
+1 -2
View File
@@ -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 };