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)); + } +}