diff --git a/README.md b/README.md index cd1a38ee..adca83d7 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,9 @@ srtla_send [OPTIONS] SRT_LISTEN_PORT SRTLA_HOST SRTLA_PORT BIND_IPS_FILE - `--no-stall-deselect`: Disable the stalled-link deselect guard (on by default). The guard skips a link whose in-flight backlog is high while its last delivery proof (an earned ACK or keepalive round-trip) has gone stale, provided a healthier link can carry the traffic. The link recovers automatically on its next keepalive round-trip, so nothing is probed blindly. This mainly helps satellite links (Starlink obstructions and handovers) that keep a large backlog while briefly delivering nothing. - `--stall-min-in-flight `: In-flight backlog (packets) at or above which a link becomes a stall candidate (default 32) - `--stall-ack-stale-ms `: Delivery-proof staleness window in milliseconds after which a stall candidate is deselected (default 3000) +- `--conn-timeout-ms `: Per-link liveness timeout; silence past this tears the link down and re-registers it (default 5000, clamped 1000–60000; runtime: `set_conn_timeout`) +- `--reconnect-fast-retry-ms `: Reconnect cadence for an uplink that was established and has timed out, during its fast-retry window (default 1000, clamped 1000–5000; runtime: `set_reconnect_fast_retry`). Matches the C reference, which re-opens the socket on every 1 s housekeeping pass, so a short modem/tether blip is re-registered before the SRT session's peer-idle timeout (5 s) ends it +- `--reconnect-fast-retry-attempts `: Failed reconnects retried at `--reconnect-fast-retry-ms` before the exponential backoff (5 s doubling, 120 s cap) takes over (default 4, max 10). `0` restores the pre-window behaviour: backoff from the first miss - `--no-rehome`: Disable whole-bond re-home (on by default). When every uplink has been down for longer than the all-links-failed window *and* the receiver hostname no longer resolves to the address the bond is pinned to, the sender moves every uplink together to the newly-resolved address and re-registers from scratch over the ordinary REG1/REG2/REG3 flow. It is deliberately conservative: a bond with any live uplink is never touched, a merely reordered DNS answer is not a move, a failed lookup is not a move, and at most one migration is attempted per minute. Pass this to keep the old behaviour of staying on the cached address until the process is restarted. - `--config `: Path to a TOML config file (reloaded on SIGHUP) - `--control-socket `: Unix domain socket path for remote control (e.g., `/tmp/srtla.sock`) diff --git a/crates/srtla-core/src/config_snapshot.rs b/crates/srtla-core/src/config_snapshot.rs index 5483a9ec..9e31aa3f 100644 --- a/crates/srtla-core/src/config_snapshot.rs +++ b/crates/srtla-core/src/config_snapshot.rs @@ -109,6 +109,23 @@ pub const CONN_TIMEOUT_MS: u64 = srtla_protocol::CONN_TIMEOUT * 1000; pub const CONN_TIMEOUT_MS_MIN: u64 = 1_000; pub const CONN_TIMEOUT_MS_MAX: u64 = 60_000; +/// Default reconnect cadence (ms) for an uplink that WAS established and has +/// timed out, during its fast-retry window. Matches the C reference +/// (`connection_housekeeping` re-opens the socket on every 1 s pass), so a +/// short interface blip is re-registered inside the SRT peer-idle timeout. +pub const RECONNECT_FAST_RETRY_MS: u64 = 1_000; +/// Clamp bounds for the fast-retry cadence. Housekeeping runs once a second, +/// so nothing below that is observable; above the exponential ladder's base +/// (5 s) the "fast" window would be slower than the ladder it precedes. +pub const RECONNECT_FAST_RETRY_MS_MIN: u64 = 1_000; +pub const RECONNECT_FAST_RETRY_MS_MAX: u64 = 5_000; +/// Default number of failed reconnect attempts retried at +/// [`RECONNECT_FAST_RETRY_MS`] before the exponential backoff ladder +/// (5 s doubling, 120 s cap) takes over. `0` disables the window: the ladder +/// applies from the first miss (the previous behaviour). +pub const RECONNECT_FAST_RETRY_ATTEMPTS: u32 = 4; +pub const RECONNECT_FAST_RETRY_ATTEMPTS_MAX: u32 = 10; + /// 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. @@ -130,6 +147,13 @@ pub struct ConfigSnapshot { /// Per-link liveness timeout in ms (default [`CONN_TIMEOUT_MS`]). Silence /// past this tears the link down and re-registers it. pub conn_timeout_ms: u64, + /// Reconnect cadence (ms) during an established link's fast-retry window + /// (default [`RECONNECT_FAST_RETRY_MS`]). + pub reconnect_fast_retry_ms: u64, + /// Failed reconnect attempts retried at `reconnect_fast_retry_ms` before + /// the exponential ladder (default [`RECONNECT_FAST_RETRY_ATTEMPTS`]; + /// `0` = ladder from the first miss). + pub reconnect_fast_retry_attempts: u32, /// One-way delivery budget in ms: the TSBPD receive delay the far-end SRT /// listener declared in its handshake response, so the deadline a packet on /// any link has to beat. Zero until the handshake crosses (and on a peer @@ -147,6 +171,8 @@ impl Default for ConfigSnapshot { stall_min_in_flight: STALL_MIN_IN_FLIGHT_PACKETS, stall_ack_stale_ms: STALL_ACK_STALE_MS, conn_timeout_ms: CONN_TIMEOUT_MS, + reconnect_fast_retry_ms: RECONNECT_FAST_RETRY_MS, + reconnect_fast_retry_attempts: RECONNECT_FAST_RETRY_ATTEMPTS, negotiated_latency_ms: 0, } } diff --git a/crates/srtla-core/src/connection/reconnection.rs b/crates/srtla-core/src/connection/reconnection.rs index 4277abd5..eec1ae57 100644 --- a/crates/srtla-core/src/connection/reconnection.rs +++ b/crates/srtla-core/src/connection/reconnection.rs @@ -1,23 +1,50 @@ use tracing::{debug, info}; use super::STARTUP_GRACE_MS; +use crate::config_snapshot::{RECONNECT_FAST_RETRY_ATTEMPTS, RECONNECT_FAST_RETRY_MS}; const BASE_RECONNECT_DELAY_MS: u64 = 5000; const MAX_BACKOFF_DELAY_MS: u64 = 120_000; const MAX_BACKOFF_COUNT: u32 = 5; /// Reconnection state and backoff tracking -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct ReconnectionState { pub last_reconnect_attempt_ms: u64, pub reconnect_failure_count: u32, pub connection_established_ms: u64, pub startup_grace_deadline_ms: u64, + /// Fast-retry window for a link that WAS established (see + /// [`RECONNECT_FAST_RETRY_MS`]). Refreshed from the runtime + /// `ConfigSnapshot` alongside `conn_timeout_ms`. + pub fast_retry_ms: u64, + pub fast_retry_attempts: u32, +} + +impl Default for ReconnectionState { + fn default() -> Self { + Self { + last_reconnect_attempt_ms: 0, + reconnect_failure_count: 0, + connection_established_ms: 0, + startup_grace_deadline_ms: 0, + fast_retry_ms: RECONNECT_FAST_RETRY_MS, + fast_retry_attempts: RECONNECT_FAST_RETRY_ATTEMPTS, + } + } } impl ReconnectionState { - /// Calculate backoff delay based on failure count. + /// Calculate backoff delay based on failure count: `fast_retry_ms` for the + /// first `fast_retry_attempts` failures (matching the C reference, which + /// re-opens the socket on every housekeeping pass), then exponential from + /// `BASE_RECONNECT_DELAY_MS`, capped at `MAX_BACKOFF_DELAY_MS`. With + /// `fast_retry_attempts == 0` this is the plain ladder. fn backoff_delay(&self) -> u64 { - let capped_failures = self.reconnect_failure_count.min(MAX_BACKOFF_COUNT); + if self.reconnect_failure_count < self.fast_retry_attempts { + return self.fast_retry_ms; + } + let capped_failures = + (self.reconnect_failure_count - self.fast_retry_attempts).min(MAX_BACKOFF_COUNT); let delay = BASE_RECONNECT_DELAY_MS.saturating_mul(1u64 << capped_failures); delay.min(MAX_BACKOFF_DELAY_MS) } @@ -58,10 +85,10 @@ impl ReconnectionState { self.reconnect_failure_count = self.reconnect_failure_count.saturating_add(1); info!( - "{}: Reconnect attempt #{}, next attempt in {}s", + "{}: Reconnect attempt #{}, next attempt in {}ms", label, self.reconnect_failure_count, - self.backoff_delay() / 1000 + self.backoff_delay() ); } @@ -76,3 +103,144 @@ impl ReconnectionState { self.startup_grace_deadline_ms = now + STARTUP_GRACE_MS; } } + +#[cfg(test)] +mod tests { + use super::*; + + /// A link that was established (REG3 seen) and has just timed out. + fn established() -> ReconnectionState { + ReconnectionState { + connection_established_ms: 1_000, + ..Default::default() + } + } + + fn established_with(fast_retry_ms: u64, fast_retry_attempts: u32) -> ReconnectionState { + ReconnectionState { + fast_retry_ms, + fast_retry_attempts, + ..established() + } + } + + /// Record an attempt at `t` and return the delay until the next one is + /// allowed, asserting the boundary is exact. + fn wait_after_attempt(state: &mut ReconnectionState, t: u64) -> u64 { + state.record_attempt("test", t); + let mut wait = 1; + while !state.should_attempt_reconnect(t + wait) { + wait += 1; + assert!(wait <= MAX_BACKOFF_DELAY_MS, "never allowed a retry"); + } + assert!(!state.should_attempt_reconnect(t + wait - 1)); + wait + } + + /// The waits between the first `n` consecutive failed attempts. + fn schedule(mut state: ReconnectionState, n: usize) -> Vec { + let mut t = 50_000; + let mut waits = Vec::new(); + for _ in 0..n { + let w = wait_after_attempt(&mut state, t); + waits.push(w); + t += w; + } + waits + } + + /// The schedule before the fast-retry window existed: `5000 << n`, capped. + const LADDER: [u64; 8] = [ + 10_000, 20_000, 40_000, 80_000, 120_000, 120_000, 120_000, 120_000, + ]; + + #[test] + fn defaults_are_one_second_for_four_attempts() { + let state = ReconnectionState::default(); + assert_eq!(state.fast_retry_ms, 1000); + assert_eq!(state.fast_retry_attempts, 4); + } + + #[test] + fn established_link_retries_after_one_second() { + let mut state = established(); + let t = 50_000; + assert!(state.should_attempt_reconnect(t)); + state.record_attempt("test", t); + assert!(!state.should_attempt_reconnect(t + 999)); + assert!(state.should_attempt_reconnect(t + 1000)); + } + + #[test] + fn fast_window_then_exponential_ladder() { + // Attempts 1-3 wait 1 s; the 4th recorded failure starts the ladder, so + // the wait before the 5th attempt is 5 s, then it doubles to the cap. + assert_eq!( + schedule(established(), 11), + vec![ + 1000, 1000, 1000, 5000, 10_000, 20_000, 40_000, 80_000, 120_000, 120_000, 120_000 + ] + ); + } + + #[test] + fn fifth_attempt_waits_five_seconds() { + let mut state = established(); + let mut t = 50_000; + for _ in 0..3 { + state.record_attempt("test", t); + t += 1000; + assert!(state.should_attempt_reconnect(t)); + } + state.record_attempt("test", t); + assert_eq!(state.reconnect_failure_count, 4); + assert!(!state.should_attempt_reconnect(t + 4999)); + assert!(state.should_attempt_reconnect(t + 5000)); + } + + #[test] + fn zero_attempts_is_the_plain_ladder() { + // Opt-out: with no fast-retry window the schedule is exactly the + // pre-existing `5000 << failure_count` ladder. + assert_eq!(schedule(established_with(1000, 0), 8), LADDER.to_vec()); + assert_eq!( + schedule(established_with(3000, 0), 8), + LADDER.to_vec(), + "the fast cadence must be irrelevant when the window is empty" + ); + } + + #[test] + fn window_is_configurable() { + assert_eq!( + schedule(established_with(2000, 2), 5), + vec![2000, 5000, 10_000, 20_000, 40_000] + ); + } + + #[test] + fn success_resets_to_fast_retry() { + let mut state = established(); + for i in 0..6 { + state.record_attempt("test", 50_000 + i); + } + state.mark_success("test"); + assert_eq!(wait_after_attempt(&mut state, 200_000), 1000); + } + + #[test] + fn initial_registration_cadence_unchanged() { + let mut state = ReconnectionState { + startup_grace_deadline_ms: 10_000, + fast_retry_ms: 5000, + fast_retry_attempts: 0, + ..Default::default() + }; + assert!(!state.should_attempt_reconnect(10_000)); + assert!(state.should_attempt_reconnect(10_001)); + state.record_attempt("test", 10_001); + assert_eq!(state.reconnect_failure_count, 0); + assert!(!state.should_attempt_reconnect(11_000)); + assert!(state.should_attempt_reconnect(11_001)); + } +} diff --git a/crates/srtla-core/src/selection/mod.rs b/crates/srtla-core/src/selection/mod.rs index 0d865aa9..4c01a26a 100644 --- a/crates/srtla-core/src/selection/mod.rs +++ b/crates/srtla-core/src/selection/mod.rs @@ -103,14 +103,16 @@ pub fn select_connection_idx( /// flag, latch, and pull are all cleared, restoring byte-for-byte baseline /// selection. /// -/// Also refreshes each link's `conn_timeout_ms` and `delay_budget_ms` from the -/// snapshot, so the runtime-tunable liveness window and the peer's declared +/// Also refreshes each link's `conn_timeout_ms`, reconnect fast-retry window +/// and `delay_budget_ms` from the snapshot, so the runtime-tunable liveness window and the peer's declared /// receive buffer reach `is_timed_out` and `update_stall_latch` callers that do /// not carry a config. #[inline] fn apply_stall_gate(conns: &mut [SrtlaConnection], current_time_ms: u64, config: &ConfigSnapshot) { for c in conns.iter_mut() { c.conn_timeout_ms = config.conn_timeout_ms; + c.reconnection.fast_retry_ms = config.reconnect_fast_retry_ms; + c.reconnection.fast_retry_attempts = config.reconnect_fast_retry_attempts; c.delay_budget_ms = config.negotiated_latency_ms; } diff --git a/src/config.rs b/src/config.rs index eb1c8b20..1e52f1cb 100644 --- a/src/config.rs +++ b/src/config.rs @@ -13,7 +13,9 @@ use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, AtomicU32, AtomicU64, O // here so the existing `crate::config::{ConfigSnapshot, STALL_*}` paths keep // resolving across the codebase. pub use srtla_core::config_snapshot::{ - CONN_TIMEOUT_MS, CONN_TIMEOUT_MS_MAX, CONN_TIMEOUT_MS_MIN, ConfigSnapshot, STALL_ACK_STALE_MS, + CONN_TIMEOUT_MS, CONN_TIMEOUT_MS_MAX, CONN_TIMEOUT_MS_MIN, ConfigSnapshot, + RECONNECT_FAST_RETRY_ATTEMPTS, RECONNECT_FAST_RETRY_ATTEMPTS_MAX, RECONNECT_FAST_RETRY_MS, + RECONNECT_FAST_RETRY_MS_MAX, RECONNECT_FAST_RETRY_MS_MIN, STALL_ACK_STALE_MS, STALL_MIN_IN_FLIGHT_PACKETS, }; use srtla_core::mode::SchedulingMode; @@ -32,6 +34,8 @@ pub struct DynamicConfig { stall_min_in_flight: Arc, stall_ack_stale_ms: Arc, conn_timeout_ms: Arc, + reconnect_fast_retry_ms: Arc, + reconnect_fast_retry_attempts: Arc, /// Whole-bond re-home on a dead bond whose receiver hostname has moved. /// /// Deliberately absent from [`ConfigSnapshot`]: that snapshot is taken on @@ -59,12 +63,15 @@ impl DynamicConfig { stall_min_in_flight: Arc::new(AtomicI32::new(STALL_MIN_IN_FLIGHT_PACKETS)), stall_ack_stale_ms: Arc::new(AtomicU64::new(STALL_ACK_STALE_MS)), conn_timeout_ms: Arc::new(AtomicU64::new(CONN_TIMEOUT_MS)), + reconnect_fast_retry_ms: Arc::new(AtomicU64::new(RECONNECT_FAST_RETRY_MS)), + reconnect_fast_retry_attempts: Arc::new(AtomicU32::new(RECONNECT_FAST_RETRY_ATTEMPTS)), rehome_on_failure: Arc::new(AtomicBool::new(true)), negotiated_latency_ms: Arc::new(AtomicU32::new(0)), } } /// Create config from CLI arguments. + #[allow(clippy::too_many_arguments)] pub fn from_cli( mode: SchedulingMode, no_quality: bool, @@ -72,6 +79,8 @@ impl DynamicConfig { stall_min_in_flight: i32, stall_ack_stale_ms: u64, conn_timeout_ms: u64, + reconnect_fast_retry_ms: u64, + reconnect_fast_retry_attempts: u32, no_rehome: bool, ) -> Self { Self { @@ -83,6 +92,12 @@ impl DynamicConfig { conn_timeout_ms: Arc::new(AtomicU64::new( conn_timeout_ms.clamp(CONN_TIMEOUT_MS_MIN, CONN_TIMEOUT_MS_MAX), )), + reconnect_fast_retry_ms: Arc::new(AtomicU64::new(clamp_fast_retry_ms( + reconnect_fast_retry_ms, + ))), + reconnect_fast_retry_attempts: Arc::new(AtomicU32::new(clamp_fast_retry_attempts( + reconnect_fast_retry_attempts, + ))), rehome_on_failure: Arc::new(AtomicBool::new(!no_rehome)), negotiated_latency_ms: Arc::new(AtomicU32::new(0)), } @@ -108,6 +123,10 @@ impl DynamicConfig { stall_min_in_flight: self.stall_min_in_flight.load(Ordering::Relaxed), stall_ack_stale_ms: self.stall_ack_stale_ms.load(Ordering::Relaxed), conn_timeout_ms: self.conn_timeout_ms.load(Ordering::Relaxed), + reconnect_fast_retry_ms: self.reconnect_fast_retry_ms.load(Ordering::Relaxed), + reconnect_fast_retry_attempts: self + .reconnect_fast_retry_attempts + .load(Ordering::Relaxed), negotiated_latency_ms: self.negotiated_latency_ms.load(Ordering::Relaxed), } } @@ -158,6 +177,34 @@ impl DynamicConfig { self.conn_timeout_ms.store(applied, Ordering::Relaxed); applied } + + /// Set the reconnect fast-retry window at runtime: the cadence (clamped to + /// [`RECONNECT_FAST_RETRY_MS_MIN`], [`RECONNECT_FAST_RETRY_MS_MAX`]) and/or + /// the attempt count (clamped to `0..=`[`RECONNECT_FAST_RETRY_ATTEMPTS_MAX`]; + /// `0` = exponential ladder from the first miss). `None` leaves a value + /// unchanged. Returns the applied `(ms, attempts)`. + pub fn set_reconnect_fast_retry(&self, ms: Option, attempts: Option) -> (u64, u32) { + if let Some(ms) = ms { + self.reconnect_fast_retry_ms + .store(clamp_fast_retry_ms(ms), Ordering::Relaxed); + } + if let Some(attempts) = attempts { + self.reconnect_fast_retry_attempts + .store(clamp_fast_retry_attempts(attempts), Ordering::Relaxed); + } + ( + self.reconnect_fast_retry_ms.load(Ordering::Relaxed), + self.reconnect_fast_retry_attempts.load(Ordering::Relaxed), + ) + } +} + +fn clamp_fast_retry_ms(ms: u64) -> u64 { + ms.clamp(RECONNECT_FAST_RETRY_MS_MIN, RECONNECT_FAST_RETRY_MS_MAX) +} + +fn clamp_fast_retry_attempts(attempts: u32) -> u32 { + attempts.min(RECONNECT_FAST_RETRY_ATTEMPTS_MAX) } /// Spawn the stdin command reader in a std::thread. Stdin on Linux @@ -202,6 +249,8 @@ mod tests { STALL_MIN_IN_FLIGHT_PACKETS, STALL_ACK_STALE_MS, CONN_TIMEOUT_MS, + RECONNECT_FAST_RETRY_MS, + RECONNECT_FAST_RETRY_ATTEMPTS, false, ); let snap = config.snapshot(); diff --git a/src/control.rs b/src/control.rs index 36b72a16..7a3aefb9 100644 --- a/src/control.rs +++ b/src/control.rs @@ -10,6 +10,8 @@ //! - `set_quality { enabled: bool }` //! - `set_stall_deselect { enabled: bool }` //! - `set_conn_timeout { ms: u64 }` (clamped; response echoes the applied value) +//! - `set_reconnect_fast_retry { ms?: u64, attempts?: u32 }` (either or both; +//! clamped; response echoes the applied pair; `attempts: 0` = plain backoff) //! - `get_status` → current `ConfigSnapshot` //! - `get_stats` → per-link telemetry //! @@ -332,6 +334,29 @@ fn handle_method( Ok(json!({ "ms": applied })) } + "set_reconnect_fast_retry" => { + let ms = params.get("ms").map(Value::as_u64); + let attempts = params + .get("attempts") + .map(|v| v.as_u64().map(|n| u32::try_from(n).unwrap_or(u32::MAX))); + if matches!(ms, Some(None)) || matches!(attempts, Some(None)) { + return Err(ErrorObject::new( + INVALID_PARAMS, + "expected params.ms: u64 and/or params.attempts: u32", + )); + } + let (ms, attempts) = (ms.flatten(), attempts.flatten()); + if ms.is_none() && attempts.is_none() { + return Err(ErrorObject::new( + INVALID_PARAMS, + "expected params.ms: u64 and/or params.attempts: u32", + )); + } + // Clamped, so echo what was applied (as set_conn_timeout does). + let (ms, attempts) = config.set_reconnect_fast_retry(ms, attempts); + Ok(json!({ "ms": ms, "attempts": attempts })) + } + "get_status" => { let snap = config.snapshot(); let (windows_received, malformed) = critical_window @@ -344,6 +369,8 @@ fn handle_method( "stall_min_in_flight": snap.stall_min_in_flight, "stall_ack_stale_ms": snap.stall_ack_stale_ms, "conn_timeout_ms": snap.conn_timeout_ms, + "reconnect_fast_retry_ms": snap.reconnect_fast_retry_ms, + "reconnect_fast_retry_attempts": snap.reconnect_fast_retry_attempts, "critical_windows_received": windows_received, "critical_malformed_datagrams": malformed, })) diff --git a/src/main.rs b/src/main.rs index 3e7fe662..6c7d5369 100644 --- a/src/main.rs +++ b/src/main.rs @@ -100,6 +100,23 @@ struct Cli { /// 1000..=60000. #[arg(long = "conn-timeout-ms", default_value_t = config::CONN_TIMEOUT_MS)] conn_timeout_ms: u64, + /// Reconnect cadence (ms) for an uplink that was established and has + /// timed out, for its first `--reconnect-fast-retry-attempts` retries + /// (the C reference retries on every 1 s housekeeping pass). A short + /// modem/tether blip is then re-registered inside the SRT peer-idle + /// timeout. Also settable at runtime (`set_reconnect_fast_retry`). + /// Clamped to 1000..=5000. + #[arg(long = "reconnect-fast-retry-ms", default_value_t = config::RECONNECT_FAST_RETRY_MS)] + reconnect_fast_retry_ms: u64, + /// Failed reconnect attempts retried at `--reconnect-fast-retry-ms` before + /// the exponential backoff (5 s doubling, 120 s cap) takes over. `0` + /// restores the previous behaviour: backoff from the first miss. + /// Clamped to 0..=10. + #[arg( + long = "reconnect-fast-retry-attempts", + default_value_t = config::RECONNECT_FAST_RETRY_ATTEMPTS + )] + reconnect_fast_retry_attempts: u32, /// Disable whole-bond re-home (on by default). When every uplink has been /// down for longer than the all-links-failed window AND the receiver @@ -174,8 +191,20 @@ async fn main() -> Result<()> { args.stall_min_in_flight, args.stall_ack_stale_ms, args.conn_timeout_ms, + args.reconnect_fast_retry_ms, + args.reconnect_fast_retry_attempts, args.no_rehome, ); + { + let snap = config.snapshot(); + tracing::info!( + "link liveness: conn_timeout_ms={} reconnect_fast_retry_ms={} \ + reconnect_fast_retry_attempts={}", + snap.conn_timeout_ms, + snap.reconnect_fast_retry_ms, + snap.reconnect_fast_retry_attempts + ); + } // Create shared stats for telemetry export let shared_stats = stats::SharedStats::new(); diff --git a/src/sender/housekeeping.rs b/src/sender/housekeeping.rs index ac174029..cd5049b8 100644 --- a/src/sender/housekeeping.rs +++ b/src/sender/housekeeping.rs @@ -343,12 +343,12 @@ mod tests { let t0 = now_ms(); // Drop all uplinks. Pin the reconnect backoff well past the whole test - // window (max failure count -> 120s backoff) so housekeeping reaches the + // window (4 fast retries + 5 ladder steps -> 120s cap) so housekeeping reaches the // timeout branch instead of attempting a socket reconnection. for conn in connections.iter_mut() { conn.mark_for_recovery(); conn.reconnection.last_reconnect_attempt_ms = t0; - conn.reconnection.reconnect_failure_count = 5; + conn.reconnection.reconnect_failure_count = 9; } // Arm: first all-down pass. Uptime is irrelevant (only now - failed_at @@ -465,7 +465,7 @@ mod tests { for conn in self.connections.iter_mut() { conn.mark_for_recovery(); conn.reconnection.last_reconnect_attempt_ms = at; - conn.reconnection.reconnect_failure_count = 5; + conn.reconnection.reconnect_failure_count = 9; } } diff --git a/src/tests/config_tests.rs b/src/tests/config_tests.rs index a136a0f5..5d29e77d 100644 --- a/src/tests/config_tests.rs +++ b/src/tests/config_tests.rs @@ -22,6 +22,8 @@ mod tests { crate::config::STALL_MIN_IN_FLIGHT_PACKETS, crate::config::STALL_ACK_STALE_MS, crate::config::CONN_TIMEOUT_MS, + crate::config::RECONNECT_FAST_RETRY_MS, + crate::config::RECONNECT_FAST_RETRY_ATTEMPTS, false, ); let snap = config.snapshot(); @@ -36,6 +38,8 @@ mod tests { crate::config::STALL_MIN_IN_FLIGHT_PACKETS, crate::config::STALL_ACK_STALE_MS, crate::config::CONN_TIMEOUT_MS, + crate::config::RECONNECT_FAST_RETRY_MS, + crate::config::RECONNECT_FAST_RETRY_ATTEMPTS, false, ); let snap = config.snapshot(); @@ -53,6 +57,85 @@ mod tests { assert_eq!(config.snapshot().conn_timeout_ms, 9_000); } + #[test] + fn test_reconnect_fast_retry_defaults_and_clamps() { + let config = DynamicConfig::new(); + let snap = config.snapshot(); + assert_eq!(snap.reconnect_fast_retry_ms, 1_000); + assert_eq!(snap.reconnect_fast_retry_attempts, 4); + + assert_eq!( + config.set_reconnect_fast_retry(Some(100), None), + (1_000, 4), + "floor" + ); + assert_eq!( + config.set_reconnect_fast_retry(Some(60_000), None), + (5_000, 4), + "ceiling" + ); + assert_eq!( + config.set_reconnect_fast_retry(None, Some(99)), + (5_000, 10), + "attempts cap" + ); + assert_eq!( + config.set_reconnect_fast_retry(Some(2_000), Some(0)), + (2_000, 0), + "opt-out" + ); + let snap = config.snapshot(); + assert_eq!(snap.reconnect_fast_retry_ms, 2_000); + assert_eq!(snap.reconnect_fast_retry_attempts, 0); + + let from_cli = DynamicConfig::from_cli( + SchedulingMode::Enhanced, + false, + false, + crate::config::STALL_MIN_IN_FLIGHT_PACKETS, + crate::config::STALL_ACK_STALE_MS, + crate::config::CONN_TIMEOUT_MS, + 200, + 50, + false, + ); + let snap = from_cli.snapshot(); + assert_eq!( + ( + snap.reconnect_fast_retry_ms, + snap.reconnect_fast_retry_attempts + ), + (1_000, 10), + "CLI values are clamped like conn_timeout_ms" + ); + } + + #[test] + fn test_set_reconnect_fast_retry_over_control() { + let config = DynamicConfig::new(); + let resp = crate::control::dispatch( + &config, + None, + None, + r#"{"jsonrpc":"2.0","id":1,"method":"set_reconnect_fast_retry","params":{"attempts":0}}"#, + ) + .unwrap() + .to_json(); + assert!(resp.contains(r#""attempts":0"#), "{resp}"); + assert!(resp.contains(r#""ms":1000"#), "{resp}"); + assert_eq!(config.snapshot().reconnect_fast_retry_attempts, 0); + + let bad = crate::control::dispatch( + &config, + None, + None, + r#"{"jsonrpc":"2.0","id":2,"method":"set_reconnect_fast_retry","params":{}}"#, + ) + .unwrap() + .to_json(); + assert!(bad.contains("-32602"), "{bad}"); + } + #[test] fn test_config_concurrent_access() { use std::thread; diff --git a/src/tests/stall_deselect_tests.rs b/src/tests/stall_deselect_tests.rs index 69da9212..719bf4dd 100644 --- a/src/tests/stall_deselect_tests.rs +++ b/src/tests/stall_deselect_tests.rs @@ -1121,4 +1121,28 @@ mod tests { "the scaled liveness window must reach is_timed_out callers" ); } + + #[test] + fn reconnect_fast_retry_reaches_the_link() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(1)); + let now = now_ms(); + assert_eq!(conns[0].reconnection.fast_retry_ms, 1_000); + assert_eq!(conns[0].reconnection.fast_retry_attempts, 4); + + let tuned = ConfigSnapshot { + reconnect_fast_retry_ms: 2_500, + reconnect_fast_retry_attempts: 0, + ..enhanced() + }; + let _ = select_connection_idx(&mut conns, None, now, &tuned); + assert_eq!( + ( + conns[0].reconnection.fast_retry_ms, + conns[0].reconnection.fast_retry_attempts + ), + (2_500, 0), + "the runtime fast-retry window must reach the reconnect logic" + ); + } } diff --git a/src/toml_config.rs b/src/toml_config.rs index 55618d6e..605b38ee 100644 --- a/src/toml_config.rs +++ b/src/toml_config.rs @@ -24,6 +24,12 @@ pub struct TomlConfig { pub stall_ack_stale_ms: u64, /// Per-link liveness timeout (ms); silence past this re-registers the link. pub conn_timeout_ms: u64, + /// Reconnect cadence (ms) for an established link's first + /// `reconnect_fast_retry_attempts` retries after it times out. + pub reconnect_fast_retry_ms: u64, + /// Failed reconnects retried at `reconnect_fast_retry_ms` before the + /// exponential backoff; `0` = backoff from the first miss. + pub reconnect_fast_retry_attempts: u32, // --- Congestion control --- /// RTT velocity threshold (ms/sample) above which window recovery is halved. @@ -57,6 +63,8 @@ impl Default for TomlConfig { stall_min_in_flight: crate::config::STALL_MIN_IN_FLIGHT_PACKETS, stall_ack_stale_ms: crate::config::STALL_ACK_STALE_MS, conn_timeout_ms: crate::config::CONN_TIMEOUT_MS, + reconnect_fast_retry_ms: crate::config::RECONNECT_FAST_RETRY_MS, + reconnect_fast_retry_attempts: crate::config::RECONNECT_FAST_RETRY_ATTEMPTS, rtt_velocity_gate: 2.0, warming_rtt_probes: 2, warming_timeout_ms: 5_000, @@ -130,10 +138,14 @@ mod tests { cooldown_duration_ms = 8000 min_switch_interval_ms = 30 switch_hysteresis = 1.20 + reconnect_fast_retry_ms = 2000 + reconnect_fast_retry_attempts = 0 "#; let cfg: TomlConfig = toml::from_str(toml_str).unwrap(); assert_eq!(cfg.mode, "classic"); assert!(cfg.no_quality); assert_eq!(cfg.warming_rtt_probes, 3); + assert_eq!(cfg.reconnect_fast_retry_ms, 2000); + assert_eq!(cfg.reconnect_fast_retry_attempts, 0); } }