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

Test conservative router recovery and recover lost receive-resume acknowledgements

Browse Source
This commit is contained in:
evgeny committed 2026-10-30 18:52:25 +01:00
1 parent 8f5882081f
commit 34cbdc3cde
2 files changed
+137 -12
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/etcp_router.c
+5 -4
View File
@@ -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);
}
}
tests/test_etcp_router_faults.c
+132 -8
View File
@@ -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()<deadline) {
if (saved_ack && get_time_tb()-last_dup >= 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<sizeof(tests)/sizeof(tests[0]); i++) {