From 34cbdc3cdea798fa706f4db80f777aeae58067c7 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 30 Oct 2026 18:52:25 +0100 Subject: [PATCH] Test conservative router recovery and recover lost receive-resume acknowledgements --- src/routing_layer/etcp_router.c | 9 +- tests/test_etcp_router_faults.c | 140 ++++++++++++++++++++++++++++++-- 2 files changed, 137 insertions(+), 12 deletions(-) diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 58a97c2c9..16ac9191a 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -837,15 +837,16 @@ static void router_retrans_timer_cb(void* arg) { } if (now >= rconn->retrans_due_tb && rconn->retrans_seq == rconn->retrans_end) { rconn->retrans_due_tb = now + router_retrans_interval(rconn); - if (!rconn->peer_receive_blocked) { + if (!rconn->peer_receive_blocked || elapsed >= router_retrans_rto(rconn)) { rconn->retrans_seq = rconn->tx_acked; - rconn->retrans_end = rconn->tx_sent; + // Если BUSY перестал обновляться, один дубликат запрашивает актуальное состояние/потерянный ACK. + rconn->retrans_end = rconn->peer_receive_blocked ? rconn->tx_acked + 1 : rconn->tx_sent; rconn->retrans_started = 0; DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, - "router retry scheduled: peer=%016llx svc=%u range=[%u,%u) rto_ms=%llu stalled_ms=%llu queued=%d", + "router retry scheduled: peer=%016llx svc=%u range=[%u,%u) rto_ms=%llu stalled_ms=%llu queued=%d busy_probe=%u", (unsigned long long)rconn->remote_node_id, rconn->svc_id, rconn->retrans_seq, rconn->retrans_end, (unsigned long long)(router_retrans_rto(rconn) / 10), - (unsigned long long)(elapsed / 10), queue_entry_count(rconn->send_q)); + (unsigned long long)(elapsed / 10), queue_entry_count(rconn->send_q), rconn->peer_receive_blocked); router_send_kick(rconn); } } diff --git a/tests/test_etcp_router_faults.c b/tests/test_etcp_router_faults.c index 918ff3524..3bda9ee85 100644 --- a/tests/test_etcp_router_faults.c +++ b/tests/test_etcp_router_faults.c @@ -157,7 +157,7 @@ static int settled(int side, int svc) { return c->tx_acked == c->tx_seq && !c->send_q->count && !c->inflight_q->count; } static int wait_settled(int side, int svc) { - uint64_t deadline = get_time_tb() + 40000; + uint64_t deadline = get_time_tb() + 80000; while (!settled(side, svc) && !failed && get_time_tb() < deadline) step(); REQUIRE(!failed); REQUIRE(settled(side, svc)); REQUIRE(ep[1-side].received[svc] == ep[side].sent[svc]); @@ -316,7 +316,7 @@ static int no_route_restore(void) { } static int timeout_with_duplicate_ack(void) { REQUIRE(setup() == 0); drop_all_ack = 1; REQUIRE(send_one(0,0) == 0); - uint64_t started = get_time_tb(), deadline = started+70000, last_dup = 0; + uint64_t started = get_time_tb(), deadline = started+100000, last_dup = 0; unsigned duplicates = 0; while (!ep[0].notified[0] && get_time_tb()= 500) { @@ -326,12 +326,12 @@ static int timeout_with_duplicate_ack(void) { step(); } REQUIRE(ep[0].notified[0] == 1); REQUIRE(ep[1].received[0] == 1); - REQUIRE(dropped_ack >= ROUTER_NO_ACK_MAX_RETRANS-1); - REQUIRE(duplicates > ROUTER_NO_ACK_MAX_RETRANS); - REQUIRE(get_time_tb() - started >= (ROUTER_NO_ACK_MAX_RETRANS - 1) * ROUTER_RETRANS_TIMEOUT_TB); + REQUIRE(dropped_ack >= 3 && dropped_ack <= 5); + REQUIRE(duplicates > 100); + REQUIRE(get_time_tb() - started >= 4 * ROUTER_RETRANS_MIN_TB); REQUIRE(ep[0].inst.router_conns->count == 0); - DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "router timeout: elapsed_tb=%llu duplicate_ACKs=%u dropped_ACKs=%d retrans_limit=%u", - (unsigned long long)(get_time_tb() - started), duplicates, dropped_ack, ROUTER_NO_ACK_MAX_RETRANS); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "router timeout: elapsed_tb=%llu duplicate_ACKs=%u dropped_ACKs=%d", + (unsigned long long)(get_time_tb() - started), duplicates, dropped_ack); cleanup(); return 0; } static int simultaneous_restart(void) { @@ -353,7 +353,7 @@ static int call_ten_second_gap(void) { while (get_time_tb() < deadline && !failed) step(); REQUIRE(!failed); REQUIRE(ep[0].notified[0] == 0); REQUIRE(rc(0,0)->reset_id == epoch && rc(0,0)->inflight_q->count == 1); - REQUIRE(rc(0,0)->c_retrans_done >= 100000 / ROUTER_RETRANS_TIMEOUT_TB); + REQUIRE(rc(0,0)->c_retrans_done == 2); // Повторы около 2с и 6с; следующая попытка около 14с. DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "router call outage: elapsed_tb=%llu retransmits=%u session=%llu inflight=%u", (unsigned long long)(get_time_tb() - started), rc(0,0)->c_retrans_done, (unsigned long long)epoch, rc(0,0)->inflight_q->count); @@ -363,6 +363,123 @@ static int call_ten_second_gap(void) { REQUIRE(rc(0,0)->reset_id == epoch && rc(0,0)->c_retrans_done > 0); cleanup(); svc_base = SVC; return 0; } + +// Дойти до передачи DATA нижнему транспорту, оставив его в wire-очереди. +static int submit_without_delivery(void) { + REQUIRE(send_one(0,0) == 0); + uint64_t deadline = get_time_tb() + 20000; + while (get_time_tb() < deadline) { + test_clock_poll(ua, 1); + if (rc(0,0)->inflight_q->count) return 0; + for (int i = 0; i < 2; i++) wire(i, ep[i].conn.send_input_q); + } + REQUIRE(0); return -1; +} + +static int slow_delivery_rto(void) { + REQUIRE(setup() == 0); + const int delays[] = {15000, 35000, 55000, 70000}; + for (unsigned i = 0; i < sizeof(delays) / sizeof(delays[0]); i++) { + REQUIRE(submit_without_delivery() == 0); + test_clock_run_for(ua, delays[i]); + REQUIRE(rc(0,0)->c_retrans_done == 0); + REQUIRE(wait_settled(0,0) == 0); + struct ETCP_ROUTER_CONN* c = rc(0,0); + REQUIRE(c->retrans_srtt_tb && c->retrans_rttvar_tb); + printf(" delay_ms=%d srtt_ms=%llu var_ms=%llu retries=%u\n", delays[i] / 10, + (unsigned long long)(c->retrans_srtt_tb / 10), (unsigned long long)(c->retrans_rttvar_tb / 10), c->c_retrans_done); + } + REQUIRE(rc(0,0)->retrans_srtt_tb > 20000); // Измерение не оборачивается на 6.55с. + cleanup(); return 0; +} + +static int idle_then_delayed_ack(void) { + REQUIRE(setup() == 0); REQUIRE(send_one(0,0) == 0); REQUIRE(wait_settled(0,0) == 0); + test_clock_run_for(ua, 100000); + REQUIRE(submit_without_delivery() == 0); + struct ETCP_ROUTER_CONN* c = rc(0,0); + test_clock_run_for(ua, 15000); + REQUIRE(!c->c_retrans_done && !c->no_ack_count && !ep[0].notified[0]); + REQUIRE(wait_settled(0,0) == 0); + REQUIRE(c->delivery_rtt_us >= 150000 && !c->c_retrans_done); + cleanup(); return 0; +} + +static int busy_receiver; +static int consumer_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* data, size_t len) { + (void)inst; (void)group; (void)peer; (void)data; (void)len; + return !busy_receiver; +} + +static int busy_then_resume(void) { + REQUIRE(setup() == 0); busy_receiver = 1; + etcp_router_set_receive_ready(&ep[1].inst, svc_base, consumer_ready); + REQUIRE(send_one(0,0) == 0); + uint64_t deadline = get_time_tb() + 120000; + while (get_time_tb() < deadline) step(); + REQUIRE(!failed && !ep[0].notified[0] && !ep[1].received[0]); + REQUIRE(rc(0,0)->peer_receive_blocked && !rc(0,0)->c_retrans_done && !rc(0,0)->no_ack_count); + busy_receiver = 0; + etcp_router_resume_receive(&ep[1].inst, 0, ep[0].inst.node_id, svc_base); + REQUIRE(wait_settled(0,0) == 0); REQUIRE(!rc(0,0)->peer_receive_blocked); + cleanup(); return 0; +} + +static int retry_waiter_backpressure(void) { + REQUIRE(setup() == 0); REQUIRE(submit_without_delivery() == 0); + struct ETCP_ROUTER_CONN* c = rc(0,0); + uint64_t srtt = c->retrans_srtt_tb; + test_clock_run_for(ua, 30000); // Исходный DATA ещё занимает нижнюю очередь, повтор только ждёт waiter. + REQUIRE(!c->c_retrans_done && !c->no_ack_count && c->send_waiter.internal); + REQUIRE(wait_settled(0,0) == 0); + REQUIRE(c->c_retrans_done <= 1); + if (c->c_retrans_done) REQUIRE(c->retrans_srtt_tb == srtt); // ACK копии не обучает RTO. + cleanup(); return 0; +} + +static int lost_resume_ack(void) { + REQUIRE(setup() == 0); busy_receiver = 1; + etcp_router_set_receive_ready(&ep[1].inst, svc_base, consumer_ready); + REQUIRE(send_one(0,0) == 0); + uint64_t deadline = get_time_tb() + 5000; + while (get_time_tb() < deadline) step(); + REQUIRE(rc(0,0)->peer_receive_blocked); + busy_receiver = 0; drop_all_ack = 1; + etcp_router_resume_receive(&ep[1].inst, 0, ep[0].inst.node_id, svc_base); + deadline = get_time_tb() + 15000; + while (get_time_tb() < deadline) step(); + REQUIRE(rc(0,0)->peer_receive_blocked && ep[1].received[0] == 1); + REQUIRE(!rc(0,0)->c_retrans_done); + drop_all_ack = 0; + REQUIRE(wait_settled(0,0) == 0); + REQUIRE(!rc(0,0)->peer_receive_blocked && rc(0,0)->c_retrans_done == 1); + cleanup(); return 0; +} + +static int own_queue_not_rtt(void) { + REQUIRE(setup() == 0); REQUIRE(send_one(0,0) == 0); REQUIRE(wait_settled(0,0) == 0); + queue_data_put(ep[0].conn.send_input_q, queue_entry_new(0)); + REQUIRE(send_one(0,0) == 0); + test_clock_run_for(ua, 80000); + struct ETCP_ROUTER_CONN* c = rc(0,0); + REQUIRE(c->send_q->count == 1 && !c->inflight_q->count && !c->retrans_timer && !c->c_retrans_done); + REQUIRE(wait_settled(0,0) == 0); + REQUIRE(c->retrans_srtt_tb < 1000 && c->delivery_rtt_us >= 1000000); + cleanup(); return 0; +} + +static int missing_route_not_attempt(void) { + REQUIRE(setup() == 0); REQUIRE(submit_without_delivery() == 0); + struct ETCP_ROUTER_CONN* c = rc(0,0); + release(queue_data_get(ep[0].conn.send_input_q)); // Потерянный пакет при разрыве нижнего транспорта. + struct conn_queue_entry* route = (struct conn_queue_entry*)ep[0].route->data; + route->conn = NULL; + test_clock_run_for(ua, 30000); + REQUIRE(c->no_route && !c->c_retrans_done && !c->no_ack_count && !ep[0].notified[0]); + route->conn = &ep[0].alternate; + REQUIRE(wait_settled(0,0) == 0); REQUIRE(c->c_retrans_done == 1); + cleanup(); return 0; +} static int size_bounds(void) { REQUIRE(setup() == 0); uint8_t* data = u_calloc(1, PKTNORM_MAX_DGRAM_SIZE); @@ -389,6 +506,13 @@ int main(void) { {"restart rejects old DATA/ACK/CLOSE/HELLO/confirm", stale_session}, {"sequence wraparound", wraparound}, {"waiter lifetime after route switch", waiter_lifetime}, {"service closes during delivery", close_from_handler}, {"route switch while old next hop blocked", route_switch}, {"no route then restore without data loss", no_route_restore}, {"duplicate ACK cannot prevent timeout", timeout_with_duplicate_ack}, + {"slow lossless path adapts RTO beyond timestamp wrap", slow_delivery_rto}, + {"idle then delayed ACK does not cause immediate retry", idle_then_delayed_ack}, + {"busy receiver pauses DATA retries and resumes", busy_then_resume}, + {"pending retry does not count as transmitted", retry_waiter_backpressure}, + {"lost resume ACK recovers with one BUSY probe", lost_resume_ack}, + {"own send queue is excluded from retransmission RTT", own_queue_not_rtt}, + {"missing route does not count as a retry attempt", missing_route_not_attempt}, {"simultaneous endpoint restart", simultaneous_restart}, {"call: 10s outage retains session and pending audio", call_ten_second_gap}, {"payload size boundaries", size_bounds}, {"producer closes during send queue drain", close_from_producer} }; for (unsigned i=0; i