From 334793f0607f0b183997a1abcddfeec9a9594de1 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 24 Sep 2026 16:04:40 +0800 Subject: [PATCH] recovery: don't let receiver traffic refresh a link awaiting REG3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A link that was established and is put into recovery (send failure, failed socket rebuild) has `connected = false` until REG3 re-admits it, and relies on `is_timed_out` to drive housekeeping's socket rebuild and REG2. If the interface returns quickly, the receiver's SRT ACK/NAKs for that address arrive on the still-open socket and refresh `last_received`; housekeeping then treats the link as alive, sends it keepalives whose replies keep refreshing it, and never re-registers it. The link carries no data (not connected) until the receiver gives up on it, which on a macOS USB tether replugged for 500 ms took 15-20 s — far past the SRT peer-idle timeout. Only refresh `last_received` on non-registration traffic for a link that is connected or has never been established, so a recovering link stays on the timeout-driven reconnect path until REG3. --- src/sender/uplink_recv.rs | 11 ++++- src/tests/mod.rs | 3 ++ src/tests/recovery_limbo_tests.rs | 79 +++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 src/tests/recovery_limbo_tests.rs diff --git a/src/sender/uplink_recv.rs b/src/sender/uplink_recv.rs index 5e022dcf..439cd09a 100644 --- a/src/sender/uplink_recv.rs +++ b/src/sender/uplink_recv.rs @@ -67,7 +67,16 @@ pub async fn process_uplink_packet( return Ok(incoming); } - conn.last_received = Some(now); + // Liveness only counts for a link that is registered, or has never been + // (initial registration keeps its own grace/timeout handling). A link + // that WAS established and is in recovery (`connected == false` until + // REG3) must stay timed out so housekeeping rebuilds its socket and + // re-sends REG2: letting the receiver's ACKs/NAKs and keepalive replies + // refresh it here parks it in limbo — alive to the timeout check, + // excluded from scheduling, never re-registered. + if conn.connected || conn.reconnection.connection_established_ms == 0 { + conn.last_received = Some(now); + } if pt == SRT_TYPE_ACK { if let Some(ack) = parse_srt_ack(data) { diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 1cf29b95..0ccc9c9a 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -27,3 +27,6 @@ pub mod integration_tests; #[cfg(test)] pub mod end_to_end_tests; + +#[cfg(test)] +pub mod recovery_limbo_tests; diff --git a/src/tests/recovery_limbo_tests.rs b/src/tests/recovery_limbo_tests.rs new file mode 100644 index 00000000..35c405f4 --- /dev/null +++ b/src/tests/recovery_limbo_tests.rs @@ -0,0 +1,79 @@ +//! A link that was established and then put into recovery must stay on the +//! timeout-driven reconnect path until REG3 re-admits it. +//! +//! Recovery (`mark_for_recovery`) clears `connected`, so the scheduler stops +//! sending data on the link, and relies on `is_timed_out` to route it through +//! housekeeping's socket rebuild + REG2. If the interface comes back quickly +//! (a replugged tether keeps its address), the receiver's traffic — SRT +//! ACK/NAKs, then replies to the keepalives housekeeping sends to a link it +//! believes alive — used to refresh `last_received`. The link then never timed +//! out, was never re-registered, and carried no data until the receiver gave up +//! on it (~15-20 s on the wire), long after the SRT peer-idle timeout. + +#[cfg(test)] +mod tests { + use srtla_core::registration::SrtlaRegistrationManager; + use srtla_core::test_helpers::create_test_connections; + use srtla_core::utils::now_ms; + use srtla_protocol::*; + + use crate::sender::process_uplink_packet; + + async fn deliver(conn: &mut srtla_core::connection::SrtlaConnection, data: &[u8]) { + let mut reg = SrtlaRegistrationManager::new(); + let listener = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let (instant_tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + process_uplink_packet(conn, 0, &mut reg, &listener, &instant_tx, None, data) + .await + .unwrap(); + } + + fn srt_ack() -> Vec { + let mut pkt = SRT_TYPE_ACK.to_be_bytes().to_vec(); + pkt.resize(44, 0); + pkt + } + + async fn recovering_link() -> srtla_core::connection::SrtlaConnection { + let mut conns = create_test_connections(1).await; + let mut conn = conns.remove(0); + conn.connected = true; + conn.reconnection.connection_established_ms = 1; + conn.last_received = Some(now_ms()); + conn.mark_for_recovery(); + assert!(!conn.connected); + assert!( + conn.is_timed_out(now_ms()), + "recovery must time the link out" + ); + conn + } + + #[tokio::test] + async fn receiver_traffic_does_not_rescue_a_recovering_link() { + let mut conn = recovering_link().await; + deliver(&mut conn, &srt_ack()).await; + assert!( + conn.is_timed_out(now_ms()), + "an SRT ACK on a link awaiting REG3 must not take it off the reconnect path" + ); + } + + #[tokio::test] + async fn keepalive_reply_does_not_rescue_a_recovering_link() { + let mut conn = recovering_link().await; + deliver(&mut conn, &create_keepalive_packet(now_ms())).await; + assert!(conn.is_timed_out(now_ms())); + } + + #[tokio::test] + async fn traffic_still_refreshes_a_connected_link() { + let mut conns = create_test_connections(1).await; + let conn = &mut conns[0]; + conn.connected = true; + conn.reconnection.connection_established_ms = 1; + conn.last_received = Some(now_ms().saturating_sub(1500)); + deliver(conn, &srt_ack()).await; + assert!(conn.last_received.unwrap() >= now_ms().saturating_sub(10)); + } +}