diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 1eb6124d..ccaac870 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -34,6 +34,7 @@ static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_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 void router_restart_send(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); @@ -49,7 +50,7 @@ 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 int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr, size_t wire_len); 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); @@ -279,6 +280,33 @@ static void router_send_rst(struct ETCP_ROUTER_CONN* rconn) { } } +// Рестарт только отправки (по RST от пира): сброс tx-состояния, очистка send_q/inflight_q. +// Приёмная сторона (rx) и сервис не трогаются. Повторные RST игнорируются через send_restart_pending. +static void router_restart_send(struct ETCP_ROUTER_CONN* rconn) { + struct UTUN_INSTANCE* inst = rconn->inst; + 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; } + { + 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); } + } + 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); + } + } + rconn->tx_seq = 0; rconn->tx_acked = 0; rconn->start_sent = 0; + rconn->send_blocked = 0; rconn->no_ack_count = 0; rconn->last_ack_changed_tb = 0; + rconn->send_restart_pending = 1; +} + static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) { static struct ETCP_CONN c; memset(&c, 0, sizeof(c)); @@ -345,6 +373,7 @@ static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { queue_resume_callback(rconn->incoming_q); queue_free(rconn->incoming_q); rconn->incoming_q = NULL; } + if (rconn->last_start_dgram) { u_free(rconn->last_start_dgram); rconn->last_start_dgram = NULL; rconn->last_start_dgram_len = 0; } } // Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/last_dgram_ts). @@ -373,7 +402,7 @@ static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { 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->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; + rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; rconn->send_restart_pending = 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; @@ -410,6 +439,7 @@ static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) { 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 (!rconn->start_sent && queue_entry_count(rconn->inflight_q) > 0) { rconn->send_blocked = 1; return; } if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { rconn->send_blocked = 1; return; @@ -468,6 +498,7 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) { if (!rconn->retrans_timer) router_retrans_schedule(rconn); } router_track_inflight_state(rconn); + if (!rconn->start_sent) { rconn->send_blocked = 1; return; } } if (queue_entry_count(rconn->send_q) == 0) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; } @@ -487,7 +518,7 @@ static void router_send_watchdog_cb(void* arg) { rconn->watchdog_timer = NULL; if (rconn->closed) return; if (queue_entry_count(rconn->send_q) == 0) return; - if (!rconn->no_route + if (!rconn->no_route && !rconn->send_blocked && 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", @@ -940,7 +971,7 @@ static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CON 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 — сессия больше не «новая» + if (rconn->tx_acked > 0) { rconn->start_sent = 1; rconn->send_restart_pending = 0; }// первый 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); @@ -1022,17 +1053,25 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* } static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, - struct SVC_ROUTE_HDR* hdr) { + struct SVC_ROUTE_HDR* hdr, size_t wire_len) { struct ETCP_ROUTER_CONN* rconn = *prconn; - if ((hdr->flags & ROUTER_FLAG_START) && rconn->rx_seq > 0) { - DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, - "router: peer restart svc_id=%u from %016llx", - 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; - rconn = *prconn; + if (!(hdr->flags & ROUTER_FLAG_START)) return 0; // не START — не рестарт + if (!rconn->peer_sync_done) return 0; // чистый: принимаем как первый пакет сеанса + + // dirty + START: дедуп по payload. Ретрансмит уже принятого init-пакета побайтно идентичен — + // это НЕ новый сеанс, рестарт не делаем (пере-ACK сделает dup-ветка в router_handle_data_packet). + if (rconn->last_start_dgram && rconn->last_start_dgram_len == wire_len + && memcmp(rconn->last_start_dgram, hdr, wire_len) == 0) { + return 0; } + + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, + "router: peer restart svc_id=%u from %016llx", + 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; + rconn = *prconn; return 0; } @@ -1041,7 +1080,17 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC 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; } + if (!rconn->peer_sync_done) { + rconn->peer_sync_done = 1; + rconn->rx_seq = seq; + // сохранить init-пакет (START) для дедупа ретрансмитов + if (((const struct SVC_ROUTE_HDR*)wire)->flags & ROUTER_FLAG_START) { + if (rconn->last_start_dgram) u_free(rconn->last_start_dgram); + rconn->last_start_dgram = u_malloc(wire_len); + if (rconn->last_start_dgram) { memcpy(rconn->last_start_dgram, wire, wire_len); rconn->last_start_dgram_len = wire_len; } + else rconn->last_start_dgram_len = 0; + } + } {// SEQ out of bounds int32_t d = (int32_t)(seq - rconn->rx_seq); @@ -1100,8 +1149,17 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) 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); + // Пир просит переинициализировать нашу отправку (его приём «чистый»). + if (rconn) { + if (rconn->send_restart_pending) { + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: RST (дубль) от %016llx svc_id=%u — игнор", + (unsigned long long)hdr->src_node_id, hdr->svc_id); + } else { + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: RST от %016llx svc_id=%u — рестарт отправки", + (unsigned long long)hdr->src_node_id, hdr->svc_id); + router_restart_send(rconn); + } + } free_entry(entry); return; } if (pl_len == 0 && (hdr->flags & ROUTER_FLAG_CLOSE)) { @@ -1114,7 +1172,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) rconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id); if (!rconn) { free_entry(entry); return; } } - if (router_check_peer_restart(inst, &rconn, hdr) != 0) { free_entry(entry); return; } + if (router_check_peer_restart(inst, &rconn, hdr, entry->len) != 0) { free_entry(entry); return; } // Сервер после рестарта «ждёт START»: пришёл не-START пакет от старой сессии → RST. if (!rconn->peer_sync_done && !(hdr->flags & ROUTER_FLAG_START)) { diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index ad231366..8169d1ff 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -121,6 +121,9 @@ struct ETCP_ROUTER_CONN { uint32_t sess_id; // локальный счётчик переустановок (0 при создании, ++ при рестарте; на wire не пишется) uint8_t start_sent; // 0 = нужно отправить START в первом data uint8_t peer_sync_done; // 0 = rx_seq ещё не синхронизирован с удалённым seq (авто-создание/рестарт) + uint8_t* last_start_dgram; // копия wire-пакета последнего принятого START (дедуп ретрансмитов) + size_t last_start_dgram_len; + uint8_t send_restart_pending; // 1 = RST принят и отправка переинициализирована (игнор повторных RST до первого ACK) // Ретрансмиты: inflight очередь struct ll_queue* inflight_q; // хеш по seq (4 байта), непрерывный блок без SACK diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index 25413fe6..633beb01 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -672,7 +672,7 @@ static int pump_data(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst, // ======================== Restart/session tests ======================== static int test_rst_flag(void) { - TEST("RST flag — local state reset, service notified"); + TEST("RST flag — send-side restart, no service notify"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; SETUP(); g_notify_count = 0; @@ -684,11 +684,13 @@ static int test_rst_flag(void) { g_notify_count = 0; inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, NULL, 0, 1, ROUTER_FLAG_RST); - if (g_notify_count != 1) FAIL("RST should notify service"); + if (g_notify_count != 0) FAIL("RST should NOT notify service"); struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); if (!c) FAIL("rconn should still exist after reset"); - if (c->tx_seq != 0 || c->rx_seq != 0 || c->start_sent != 0) FAIL("rconn not reset by RST"); + if (c->tx_seq != 0 || c->start_sent != 0) FAIL("send side not reset by RST"); + if (c->rx_seq != (uint32_t)n) FAIL("rx side should be untouched by RST"); + if (!c->send_restart_pending) FAIL("send_restart_pending not set by RST"); etcp_router_conn_close(c); TEARDOWN(); @@ -696,6 +698,29 @@ static int test_rst_flag(void) { return 0; } +static int test_start_retransmit_no_restart(void) { + TEST("START retransmit (same payload) — no restart"); + struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; + SETUP(); + g_notify_count = 0; + + int n = rand() % 10 + 1; + rx_reset(0xC1); + pump_data(recv_cb, &inst, 0xC1, n, 0); + if (rx.delivered != n) FAIL("initial pump failed"); + + g_notify_count = 0; + // ретрансмит первого пакета (seq=0, START, тот же payload) — не должен рестартовать + uint8_t data[] = { TEST_SVC_ID, 0xC1, 0 }; + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START); + if (g_notify_count != 0) FAIL("START retransmit should NOT restart"); + if (rx.delivered != n) FAIL("duplicate START should not deliver"); + + TEARDOWN(); + PASS(); + return 0; +} + static int test_client_restart(void) { TEST("client restart — START flag detected after random data"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; @@ -1157,6 +1182,7 @@ int main(void) { test_timestamp(); test_conn_send_no_bgp(); test_rst_flag(); + test_start_retransmit_no_restart(); test_client_restart(); test_sess_id_increment(); test_server_reinit();