Refactor SRTLA components and enhance stall gate functionality

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["crates/network-sim"]
+members = ["crates/network-sim", "crates/srtla-core", "crates/srtla-protocol"]
 
 [package]
 resolver = "3"
@@ -17,6 +17,8 @@ categories = ["network-programming", "multimedia"]
 rust-version = "1.88"
 
 [dependencies]
+srtla-core = { path = "crates/srtla-core" }
+srtla-protocol = { path = "crates/srtla-protocol" }
 anyhow = "1.0"
 clap = { version = "4.5", features = ["derive"] }
 rand = "0.9"
@@ -50,8 +52,9 @@ libc = "0.2"
 # test_helpers exposes advance_test_clock(), which calls tokio::time::advance —
 # only available with tokio's test-util. Pull it here so the test-internals
 # surface compiles standalone (e.g. `cargo build --all-features`), not just under
-# `cargo test` where dev-deps happen to unify test-util in.
-test-internals = ["tokio/test-util"]
+# `cargo test` where dev-deps happen to unify test-util in. Forward to srtla-core
+# so its cross-crate test surface (public fields, construction helpers) is exposed.
+test-internals = ["srtla-core/test-internals", "tokio/test-util"]
 
 [lib]
 name = "srtla_send"
@@ -62,6 +65,9 @@ name = "srtla_send"
 path = "src/main.rs"
 
 [dev-dependencies]
+# Enable srtla-core's test-internals surface for our own cross-crate tests
+# (public struct fields, construction helpers, pub(crate) predicates).
+srtla-core = { path = "crates/srtla-core", features = ["test-internals"] }
 tokio-test = "0.4"
 tempfile = "3"
 assert_matches = "1"

crates/srtla-core/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "srtla-core"
+version = "3.0.0"
+edition = "2024"
+description = "Pure sans-IO core for the SRTLA sender: connection state, path selection, registration, congestion control"
+license = "MIT"
+authors = ["Thomas Lekanger <mail@datagutt.no>"]
+repository = "https://github.com/irlserver/srtla_send"
+
+[dependencies]
+srtla-protocol = { path = "../srtla-protocol" }
+rand = "0.9"
+rustc-hash = "2.1"
+smallvec = "=2.0.0-alpha.12"
+tracing = "0.1"
+# Only pulled in by the test-internals feature: advance_test_clock() in the core
+# test-helpers surface calls tokio::time::advance, which must compile even in a
+# non-test build when a downstream crate enables the feature for its own tests.
+tokio = { version = "1.49", default-features = false, optional = true }
+
+[dev-dependencies]
+# The inline module tests spin up a small runtime; separate from the optional
+# dep above so `cargo test -p srtla-core` always has rt + macros available.
+tokio = { version = "1.49", features = ["rt-multi-thread", "macros", "time", "test-util"] }
+
+[features]
+# Exposes the test-only surface (public struct fields, construction helpers,
+# pub(crate) predicates) so the parent srtla_send crate can drive core types
+# from its own cross-crate tests.
+test-internals = ["dep:tokio", "tokio/time", "tokio/test-util"]

crates/srtla-core/src/config_snapshot.rs

@@ -0,0 +1,89 @@
+//! Pure, hot-path configuration snapshot (core).
+//!
+//! Split out of `config` so the scheduler core can consume a config view without
+//! the shell coupling that `DynamicConfig` carries (the control dispatcher and
+//! `SharedStats`). The live `DynamicConfig::snapshot()` builds one of these once
+//! per select iteration; this type depends only on `SchedulingMode`.
+
+use crate::mode::SchedulingMode;
+
+/// In-flight packet backlog at or above which a link is a stall candidate
+/// under the `stall_deselect` guard (default on).
+pub const STALL_MIN_IN_FLIGHT_PACKETS: i32 = 32;
+
+/// Ceiling (ms) on the delivery-proof staleness window under
+/// `stall_deselect`. The effective window is RTT-adaptive —
+/// `clamp(STALL_STALE_RTT_MULT x smoothed RTT, STALL_STALE_FLOOR_MS, this)` —
+/// so a fast link is pulled within hundreds of ms of going dark instead of
+/// always waiting the full ceiling; the reaction window is exactly the run of
+/// packets that will arrive late (reorder holes) at the receiver. A link with
+/// no RTT baseline yet falls back to the ceiling. Kept well below
+/// `CONN_TIMEOUT` (15 s): deselect is a selection penalty ONLY, never a
+/// liveness/timeout shortcut.
+pub const STALL_ACK_STALE_MS: u64 = 3000;
+
+/// Multiplier on smoothed RTT for the adaptive staleness window. Delivery
+/// proof on a loaded link normally arrives every RTT (earned SRTLA ACKs), so
+/// four missed round-trips is a strong stall signal without tripping on a
+/// single lost ACK.
+pub const STALL_STALE_RTT_MULT: u64 = 4;
+
+/// Floor (ms) on the adaptive staleness window. Sits above the 400-800 ms
+/// HARQ stalls that are routine on bonded cellular, so a normal
+/// retransmission pause never gates a link.
+pub const STALL_STALE_FLOOR_MS: u64 = 1000;
+
+/// Rejoin dwell as a multiple of the effective staleness window. A gated link
+/// must hold continuous fresh delivery proof this long before it rejoins the
+/// rotation: quick to drop, conservative to rejoin, so a still-marginal link
+/// cannot flap back in and re-glitch the stream.
+pub const STALL_REJOIN_DWELL_MULT: u64 = 2;
+
+/// Duplicate-probe rate on a stall-gated link: one copy of every Nth routed
+/// data packet is also sent on each gated link. The copies are redundant (the
+/// SRT receiver dedups by sequence number), so a late or lost probe cannot
+/// stall the receiver buffer, but a delivered one earns the link an SRTLA ACK
+/// — real data-sized delivery proof, where keepalives alone only prove the
+/// path echoes 38-byte control frames. ~1% overhead at the default.
+pub const STALL_PROBE_ONE_IN_N: u32 = 100;
+
+/// Snapshot of configuration for efficient hot-path access.
+/// Call `DynamicConfig::snapshot()` once per select iteration to avoid
+/// multiple atomic loads per packet in the hot path.
+#[derive(Clone, Copy, Debug)]
+pub struct ConfigSnapshot {
+    pub mode: SchedulingMode,
+    pub quality_enabled: bool,
+    /// Stalled-link deselect (default ON). On, the selection layer excludes a
+    /// link whose in-flight backlog is high while its last delivery proof has
+    /// gone stale, provided at least one healthier link can carry the traffic.
+    /// Off (`--no-stall-deselect`), selection is byte-for-byte unchanged.
+    pub stall_deselect: bool,
+    /// In-flight threshold for `stall_deselect` (default [`STALL_MIN_IN_FLIGHT_PACKETS`]).
+    pub stall_min_in_flight: i32,
+    /// Ceiling in ms on the RTT-adaptive delivery-proof staleness window for
+    /// `stall_deselect` (default [`STALL_ACK_STALE_MS`]; see
+    /// [`SrtlaConnection::effective_stall_stale_ms`](crate::connection::SrtlaConnection::effective_stall_stale_ms)).
+    pub stall_ack_stale_ms: u64,
+}
+
+impl Default for ConfigSnapshot {
+    fn default() -> Self {
+        Self {
+            mode: SchedulingMode::Enhanced,
+            quality_enabled: true,
+            stall_deselect: true,
+            stall_min_in_flight: STALL_MIN_IN_FLIGHT_PACKETS,
+            stall_ack_stale_ms: STALL_ACK_STALE_MS,
+        }
+    }
+}
+
+impl ConfigSnapshot {
+    /// Check if quality scoring is effective for the current mode.
+    /// Quality scoring only applies to enhanced mode.
+    #[inline]
+    pub fn effective_quality_enabled(&self) -> bool {
+        self.quality_enabled && !self.mode.is_classic()
+    }
+}

src/connection/ack_nak.rs → crates/srtla-core/src/connection/ack_nak.rs

@@ -1,7 +1,8 @@
 use std::cmp::min;
 
+use srtla_protocol::*;
+
 use super::SrtlaConnection;
-use crate::protocol::*;
 
 impl SrtlaConnection {
     /// Register a packet as in-flight. O(1) insert.

src/connection/batch_send.rs → crates/srtla-core/src/connection/batch_send.rs

@@ -33,13 +33,18 @@
 //! The `set_regime` setter is called from `housekeeping` based on each
 //! connection's `current_bitrate_bps` snapshot.
 
-use std::sync::Arc;
-
 use smallvec::SmallVec;
-use tokio::time::Instant;
 use tracing::debug;
 
-use super::batch_recv::{BATCH_SEND_SIZE, BatchUdpSocket};
+/// Max datagrams per `sendmmsg` batch. Defined here (core) so the pure
+/// `drain()` return type and the shell's `net::send_all_datagrams` chunking
+/// agree on one value without core depending on the socket layer.
+pub const BATCH_SEND_SIZE: usize = 32;
+
+/// One drained datagram awaiting transmission: `(bytes, seq, queue_time_ms)`.
+/// `seq` is `Some` for tracked data packets and `None` for untracked ones; the
+/// shell registers the tracked ones for in-flight accounting after sending.
+pub type DrainedPacket = (SmallVec<u8, 1500>, Option<u32>, u64);
 
 /// Bitrate above which a connection is treated as high-load.
 pub const HIGH_LOAD_THRESHOLD_BPS: f64 = 5_000_000.0;
@@ -122,8 +127,10 @@ pub struct BatchSender {
     /// Timestamps when packets were queued (parallel to queue)
     queue_times: Vec<u64>,
 
-    /// Last time the queue was flushed
-    last_flush_time: Instant,
+    /// `now_ms()` of the last drain. Drives the 15ms time-flush window against
+    /// the injected monotonic clock — no wall/`Instant` clock lives here so the
+    /// queue is pure, testable state and the actual send lifts to the shell.
+    last_flush_ms: u64,
 
     /// Current batch regime. Updated by housekeeping when the
     /// connection's bitrate crosses a threshold.
@@ -143,7 +150,7 @@ impl BatchSender {
             queue: Vec::with_capacity(BATCH_SIZE_HIGH_LOAD),
             sequences: Vec::with_capacity(BATCH_SIZE_HIGH_LOAD),
             queue_times: Vec::with_capacity(BATCH_SIZE_HIGH_LOAD),
-            last_flush_time: Instant::now(),
+            last_flush_ms: 0,
             regime: BatchRegime::default(),
         }
     }
@@ -174,11 +181,13 @@ impl BatchSender {
         self.regime
     }
 
-    /// Check if the queue needs flushing based on time
+    /// Check if the queue needs flushing based on time.
+    ///
+    /// `now_ms` is the injected monotonic clock; a drain is due once the queue
+    /// has held packets for at least [`FLUSH_INTERVAL_MS`].
     #[inline]
-    pub fn needs_time_flush(&self) -> bool {
-        !self.queue.is_empty()
-            && self.last_flush_time.elapsed().as_millis() >= FLUSH_INTERVAL_MS as u128
+    pub fn needs_time_flush(&self, now_ms: u64) -> bool {
+        !self.queue.is_empty() && now_ms.saturating_sub(self.last_flush_ms) >= FLUSH_INTERVAL_MS
     }
 
     /// Check if there are any packets queued
@@ -195,89 +204,40 @@ impl BatchSender {
         self.queue.len() as i32
     }
 
-    /// Flush all queued packets to the socket
+    /// Drain the queue, returning every queued datagram with its tracking info.
     ///
-    /// Returns a vector of (seq, queue_time) pairs for packets that need tracking.
-    /// The caller should update in-flight tracking based on these.
-    pub async fn flush(
-        &mut self,
-        socket: &Arc<BatchUdpSocket>,
-    ) -> std::io::Result<Vec<(Option<u32>, u64)>> {
+    /// Pure: it performs no I/O. The queue is emptied and the flush window is
+    /// re-armed at `now_ms`; the shell transmits the returned bytes (see
+    /// [`send_all_datagrams`]) and registers the `Some(seq)` entries for
+    /// in-flight tracking. Returns empty when nothing was queued.
+    pub fn drain(&mut self, now_ms: u64) -> SmallVec<DrainedPacket, BATCH_SEND_SIZE> {
+        self.last_flush_ms = now_ms;
         if self.queue.is_empty() {
-            return Ok(Vec::new());
+            return SmallVec::new();
         }
 
         let packet_count = self.queue.len();
-        let mut sent_count = 0;
-
-        // One `sendmmsg` per BATCH_SEND_SIZE datagrams. The kernel may accept
-        // fewer than offered (short send) once the socket buffer fills, so loop
-        // until the queue is drained rather than assuming a full batch left.
-        while sent_count < packet_count {
-            let take = (packet_count - sent_count).min(BATCH_SEND_SIZE);
-
-            // Scoped so the borrow of `self.queue` ends before the error path
-            // below mutates it.
-            let result = {
-                let mut bufs: SmallVec<&[u8], BATCH_SEND_SIZE> = SmallVec::new();
-                for packet in &self.queue[sent_count..sent_count + take] {
-                    bufs.push(&packet[..]);
-                }
-                socket.send_batch(&bufs).await
-            };
-
-            match result {
-                // Ok(0) would spin forever; treat a no-progress send as an error
-                // so the link is retried rather than livelocked.
-                Ok(0) => {
-                    self.queue.drain(..sent_count);
-                    self.sequences.drain(..sent_count);
-                    self.queue_times.drain(..sent_count);
-                    self.last_flush_time = Instant::now();
-                    return Err(std::io::Error::new(
-                        std::io::ErrorKind::WriteZero,
-                        "sendmmsg accepted no datagrams",
-                    ));
-                }
-                Ok(n) => sent_count += n,
-                Err(e) => {
-                    // Partial failure: remove already-sent packets to avoid duplicates
-                    self.queue.drain(..sent_count);
-                    self.sequences.drain(..sent_count);
-                    self.queue_times.drain(..sent_count);
-                    self.last_flush_time = Instant::now();
-                    return Err(e);
-                }
-            }
-        }
-
-        // Collect tracking info before clearing
-        let tracking_info: Vec<(Option<u32>, u64)> = self
-            .sequences
-            .iter()
-            .zip(self.queue_times.iter())
-            .map(|(&seq, &time)| (seq, time))
+        let out: SmallVec<DrainedPacket, BATCH_SEND_SIZE> = self
+            .queue
+            .drain(..)
+            .zip(self.sequences.drain(..))
+            .zip(self.queue_times.drain(..))
+            .map(|((data, seq), time)| (data, seq, time))
             .collect();
 
-        // Clear the queue
-        self.queue.clear();
-        self.sequences.clear();
-        self.queue_times.clear();
-        self.last_flush_time = Instant::now();
-
         if packet_count > 1 {
-            debug!("Batch flush: sent {} packets in one batch", packet_count);
+            debug!("Batch drain: {} packets ready to send", packet_count);
         }
 
-        Ok(tracking_info)
+        out
     }
 
     /// Reset the batch sender state (for reconnection)
     pub fn reset(&mut self) {
         self.queue.clear();
         self.sequences.clear();
         self.queue_times.clear();
-        self.last_flush_time = Instant::now();
+        self.last_flush_ms = 0;
     }
 }
 
@@ -307,11 +267,12 @@ mod tests {
         let data = [0u8; 100];
 
         sender.queue_packet(&data, Some(1), 0);
-        assert!(!sender.needs_time_flush()); // Just queued, shouldn't need flush
+        assert!(!sender.needs_time_flush(0)); // Just queued at t=0, shouldn't need flush
+        assert!(!sender.needs_time_flush(FLUSH_INTERVAL_MS - 1)); // Under the window
 
-        // Simulate time passing
-        sender.last_flush_time = Instant::now() - std::time::Duration::from_millis(20);
-        assert!(sender.needs_time_flush()); // Now should need flush
+        // Once FLUSH_INTERVAL_MS has elapsed against the injected clock.
+        assert!(sender.needs_time_flush(FLUSH_INTERVAL_MS));
+        assert!(sender.needs_time_flush(20));
     }
 
     #[test]