From 67d7c46e61702fef8bf62048969f43f3708a5ec6 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:01:22 +0100 Subject: [PATCH 01/21] refactor(core): add state machine transition contracts --- crates/slipstream-core/src/lib.rs | 1 + crates/slipstream-core/src/state_machine.rs | 357 ++++++++++++++++++++ 2 files changed, 358 insertions(+) create mode 100644 crates/slipstream-core/src/state_machine.rs diff --git a/crates/slipstream-core/src/lib.rs b/crates/slipstream-core/src/lib.rs index 61898337..abbd7801 100644 --- a/crates/slipstream-core/src/lib.rs +++ b/crates/slipstream-core/src/lib.rs @@ -5,6 +5,7 @@ pub mod invariants; mod macros; pub mod net; pub mod sip003; +pub mod state_machine; pub mod stream; pub mod tcp; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs}; diff --git a/crates/slipstream-core/src/state_machine.rs b/crates/slipstream-core/src/state_machine.rs new file mode 100644 index 00000000..6e8a6eb5 --- /dev/null +++ b/crates/slipstream-core/src/state_machine.rs @@ -0,0 +1,357 @@ +use std::fmt; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConnState { + Init, + Handshaking, + Active, + Degraded, + Recovering, + Draining, + Closed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConnEvent { + StartHandshake, + HandshakeReady, + HandshakeFailed, + Degrade, + Recover, + Drain, + Close, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResolverHealthState { + Unknown, + Probing, + Healthy, + Degraded, + Cooling, + Dead, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResolverRole { + Active, + Standby, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResolverEvent { + ProbeStarted, + ProbeSucceeded, + ProbeFailed, + MarkDegraded, + StartCooldown, + CooldownExpired, + MarkDead, + Revive, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StreamState { + Idle, + Open, + FinLocal, + FinRemote, + Closed, + Reset, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StreamEvent { + Open, + LocalFin, + RemoteFin, + AckClose, + Reset, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MachineKind { + Connection, + Resolver, + Stream, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TransitionError { + pub machine: MachineKind, + pub state: &'static str, + pub event: &'static str, +} + +impl TransitionError { + fn new(machine: MachineKind, state: &'static str, event: &'static str) -> Self { + Self { + machine, + state, + event, + } + } +} + +impl fmt::Display for TransitionError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "invalid {:?} transition: state={} event={}", + self.machine, self.state, self.event + ) + } +} + +impl std::error::Error for TransitionError {} + +pub fn validate_conn_transition( + state: ConnState, + event: ConnEvent, +) -> Result { + if event == ConnEvent::Close { + return match state { + ConnState::Closed => Err(TransitionError::new( + MachineKind::Connection, + conn_state_name(state), + conn_event_name(event), + )), + _ => Ok(ConnState::Closed), + }; + } + + let next = match (state, event) { + (ConnState::Init, ConnEvent::StartHandshake) => ConnState::Handshaking, + (ConnState::Handshaking, ConnEvent::HandshakeReady) => ConnState::Active, + (ConnState::Handshaking, ConnEvent::HandshakeFailed) => ConnState::Recovering, + (ConnState::Active, ConnEvent::Degrade) => ConnState::Degraded, + (ConnState::Active, ConnEvent::Drain) => ConnState::Draining, + (ConnState::Degraded, ConnEvent::Recover) => ConnState::Active, + (ConnState::Degraded, ConnEvent::Drain) => ConnState::Draining, + (ConnState::Recovering, ConnEvent::StartHandshake) => ConnState::Handshaking, + (ConnState::Recovering, ConnEvent::Drain) => ConnState::Draining, + (ConnState::Draining, ConnEvent::Drain) => ConnState::Draining, + _ => { + return Err(TransitionError::new( + MachineKind::Connection, + conn_state_name(state), + conn_event_name(event), + )); + } + }; + + Ok(next) +} + +pub fn validate_resolver_transition( + state: ResolverHealthState, + event: ResolverEvent, +) -> Result { + let next = match (state, event) { + (ResolverHealthState::Unknown, ResolverEvent::ProbeStarted) => ResolverHealthState::Probing, + (ResolverHealthState::Probing, ResolverEvent::ProbeSucceeded) => { + ResolverHealthState::Healthy + } + (ResolverHealthState::Probing, ResolverEvent::ProbeFailed) => ResolverHealthState::Degraded, + (ResolverHealthState::Healthy, ResolverEvent::MarkDegraded) => { + ResolverHealthState::Degraded + } + (ResolverHealthState::Healthy, ResolverEvent::MarkDead) => ResolverHealthState::Dead, + (ResolverHealthState::Degraded, ResolverEvent::ProbeSucceeded) => { + ResolverHealthState::Healthy + } + (ResolverHealthState::Degraded, ResolverEvent::StartCooldown) => { + ResolverHealthState::Cooling + } + (ResolverHealthState::Degraded, ResolverEvent::MarkDead) => ResolverHealthState::Dead, + (ResolverHealthState::Cooling, ResolverEvent::CooldownExpired) => { + ResolverHealthState::Probing + } + (ResolverHealthState::Cooling, ResolverEvent::MarkDead) => ResolverHealthState::Dead, + (ResolverHealthState::Dead, ResolverEvent::Revive) => ResolverHealthState::Probing, + _ => { + return Err(TransitionError::new( + MachineKind::Resolver, + resolver_state_name(state), + resolver_event_name(event), + )); + } + }; + + Ok(next) +} + +pub fn validate_stream_transition( + state: StreamState, + event: StreamEvent, +) -> Result { + let next = match (state, event) { + (StreamState::Idle, StreamEvent::Open) => StreamState::Open, + (StreamState::Open, StreamEvent::LocalFin) => StreamState::FinLocal, + (StreamState::Open, StreamEvent::RemoteFin) => StreamState::FinRemote, + (StreamState::Open, StreamEvent::Reset) => StreamState::Reset, + (StreamState::FinLocal, StreamEvent::RemoteFin) => StreamState::Closed, + (StreamState::FinLocal, StreamEvent::AckClose) => StreamState::Closed, + (StreamState::FinLocal, StreamEvent::Reset) => StreamState::Reset, + (StreamState::FinRemote, StreamEvent::LocalFin) => StreamState::Closed, + (StreamState::FinRemote, StreamEvent::AckClose) => StreamState::Closed, + (StreamState::FinRemote, StreamEvent::Reset) => StreamState::Reset, + (StreamState::Reset, StreamEvent::AckClose) => StreamState::Closed, + _ => { + return Err(TransitionError::new( + MachineKind::Stream, + stream_state_name(state), + stream_event_name(event), + )); + } + }; + + Ok(next) +} + +fn conn_state_name(state: ConnState) -> &'static str { + match state { + ConnState::Init => "Init", + ConnState::Handshaking => "Handshaking", + ConnState::Active => "Active", + ConnState::Degraded => "Degraded", + ConnState::Recovering => "Recovering", + ConnState::Draining => "Draining", + ConnState::Closed => "Closed", + } +} + +fn conn_event_name(event: ConnEvent) -> &'static str { + match event { + ConnEvent::StartHandshake => "StartHandshake", + ConnEvent::HandshakeReady => "HandshakeReady", + ConnEvent::HandshakeFailed => "HandshakeFailed", + ConnEvent::Degrade => "Degrade", + ConnEvent::Recover => "Recover", + ConnEvent::Drain => "Drain", + ConnEvent::Close => "Close", + } +} + +fn resolver_state_name(state: ResolverHealthState) -> &'static str { + match state { + ResolverHealthState::Unknown => "Unknown", + ResolverHealthState::Probing => "Probing", + ResolverHealthState::Healthy => "Healthy", + ResolverHealthState::Degraded => "Degraded", + ResolverHealthState::Cooling => "Cooling", + ResolverHealthState::Dead => "Dead", + } +} + +fn resolver_event_name(event: ResolverEvent) -> &'static str { + match event { + ResolverEvent::ProbeStarted => "ProbeStarted", + ResolverEvent::ProbeSucceeded => "ProbeSucceeded", + ResolverEvent::ProbeFailed => "ProbeFailed", + ResolverEvent::MarkDegraded => "MarkDegraded", + ResolverEvent::StartCooldown => "StartCooldown", + ResolverEvent::CooldownExpired => "CooldownExpired", + ResolverEvent::MarkDead => "MarkDead", + ResolverEvent::Revive => "Revive", + } +} + +fn stream_state_name(state: StreamState) -> &'static str { + match state { + StreamState::Idle => "Idle", + StreamState::Open => "Open", + StreamState::FinLocal => "FinLocal", + StreamState::FinRemote => "FinRemote", + StreamState::Closed => "Closed", + StreamState::Reset => "Reset", + } +} + +fn stream_event_name(event: StreamEvent) -> &'static str { + match event { + StreamEvent::Open => "Open", + StreamEvent::LocalFin => "LocalFin", + StreamEvent::RemoteFin => "RemoteFin", + StreamEvent::AckClose => "AckClose", + StreamEvent::Reset => "Reset", + } +} + +#[cfg(test)] +mod tests { + use super::{ + validate_conn_transition, validate_resolver_transition, validate_stream_transition, + ConnEvent, ConnState, MachineKind, ResolverEvent, ResolverHealthState, StreamEvent, + StreamState, + }; + + #[test] + fn connection_transition_happy_path() { + let state = validate_conn_transition(ConnState::Init, ConnEvent::StartHandshake) + .expect("start handshake should succeed"); + let state = validate_conn_transition(state, ConnEvent::HandshakeReady) + .expect("handshake ready should succeed"); + let state = + validate_conn_transition(state, ConnEvent::Degrade).expect("degrade should succeed"); + let state = + validate_conn_transition(state, ConnEvent::Recover).expect("recover should succeed"); + let state = + validate_conn_transition(state, ConnEvent::Drain).expect("drain should succeed"); + let state = + validate_conn_transition(state, ConnEvent::Close).expect("close should succeed"); + assert_eq!(state, ConnState::Closed); + } + + #[test] + fn rejects_connection_invalid_transition_with_context() { + let err = validate_conn_transition(ConnState::Init, ConnEvent::Recover) + .expect_err("recover from init should fail"); + assert_eq!(err.machine, MachineKind::Connection); + assert_eq!(err.state, "Init"); + assert_eq!(err.event, "Recover"); + } + + #[test] + fn resolver_transition_happy_path() { + let state = + validate_resolver_transition(ResolverHealthState::Unknown, ResolverEvent::ProbeStarted) + .expect("probe start should succeed"); + let state = validate_resolver_transition(state, ResolverEvent::ProbeSucceeded) + .expect("probe success should succeed"); + let state = validate_resolver_transition(state, ResolverEvent::MarkDegraded) + .expect("degrade should succeed"); + let state = validate_resolver_transition(state, ResolverEvent::StartCooldown) + .expect("start cooldown should succeed"); + let state = validate_resolver_transition(state, ResolverEvent::CooldownExpired) + .expect("cooldown expiry should succeed"); + assert_eq!(state, ResolverHealthState::Probing); + } + + #[test] + fn resolver_dead_can_revive_to_probing() { + let state = validate_resolver_transition(ResolverHealthState::Dead, ResolverEvent::Revive) + .expect("revive should succeed"); + assert_eq!(state, ResolverHealthState::Probing); + } + + #[test] + fn stream_transition_happy_path() { + let state = validate_stream_transition(StreamState::Idle, StreamEvent::Open) + .expect("open should succeed"); + let state = validate_stream_transition(state, StreamEvent::LocalFin) + .expect("local fin should succeed"); + let state = validate_stream_transition(state, StreamEvent::RemoteFin) + .expect("remote fin should succeed"); + assert_eq!(state, StreamState::Closed); + } + + #[test] + fn stream_invalid_transition_reports_context() { + let err = validate_stream_transition(StreamState::Idle, StreamEvent::RemoteFin) + .expect_err("remote fin from idle should fail"); + assert_eq!(err.machine, MachineKind::Stream); + assert_eq!(err.state, "Idle"); + assert_eq!(err.event, "RemoteFin"); + } +} From edccaf81a9b4345ecf6a33bf8585c7c4da5f1de8 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:04:42 +0100 Subject: [PATCH 02/21] feat(client): add resolver telemetry and switch reason taxonomy --- crates/slipstream-client/src/dns.rs | 2 +- crates/slipstream-client/src/dns/debug.rs | 194 ++++++++++++++++--- crates/slipstream-client/src/dns/path.rs | 3 + crates/slipstream-client/src/dns/poll.rs | 6 +- crates/slipstream-client/src/dns/resolver.rs | 11 +- crates/slipstream-client/src/runtime.rs | 31 ++- 6 files changed, 208 insertions(+), 39 deletions(-) diff --git a/crates/slipstream-client/src/dns.rs b/crates/slipstream-client/src/dns.rs index 0b077b32..c7369266 100644 --- a/crates/slipstream-client/src/dns.rs +++ b/crates/slipstream-client/src/dns.rs @@ -4,7 +4,7 @@ mod poll; mod resolver; mod response; -pub(crate) use debug::maybe_report_debug; +pub(crate) use debug::{maybe_report_debug, record_resolver_switch, ResolverSwitchReason}; pub(crate) use path::{add_paths, refresh_resolver_path, resolver_mode_to_c}; pub(crate) use poll::{expire_inflight_polls, send_poll_queries}; pub(crate) use resolver::{ diff --git a/crates/slipstream-client/src/dns/debug.rs b/crates/slipstream-client/src/dns/debug.rs index 4b60a20f..417606a5 100644 --- a/crates/slipstream-client/src/dns/debug.rs +++ b/crates/slipstream-client/src/dns/debug.rs @@ -1,5 +1,5 @@ use crate::pacing::PacingBudgetSnapshot; -use tracing::debug; +use tracing::{debug, info}; use super::resolver::ResolverState; @@ -25,6 +25,15 @@ pub(crate) struct DebugMetrics { pub(crate) last_report_send_packets: u64, pub(crate) last_report_send_bytes: u64, pub(crate) last_report_polls: u64, + pub(crate) inflight_poll_timeouts: u64, + pub(crate) path_probe_successes: u64, + pub(crate) path_probe_failures: u64, + pub(crate) path_rtt_us: u64, + pub(crate) path_cwnd: u64, + pub(crate) path_bytes_in_transit: u64, + pub(crate) path_pacing_rate: u64, + pub(crate) switch_to_count: u64, + pub(crate) switch_from_count: u64, } impl DebugMetrics { @@ -49,10 +58,82 @@ impl DebugMetrics { last_report_send_packets: 0, last_report_send_bytes: 0, last_report_polls: 0, + inflight_poll_timeouts: 0, + path_probe_successes: 0, + path_probe_failures: 0, + path_rtt_us: 0, + path_cwnd: 0, + path_bytes_in_transit: 0, + path_pacing_rate: 0, + switch_to_count: 0, + switch_from_count: 0, } } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ResolverSwitchReason { + StartupPrimary, + ManualOverride, + ProbeRecovery, + TimeoutStreakExceeded, + LossSpike, + LatencyRegression, + PathUnavailable, + CooldownExpired, +} + +impl ResolverSwitchReason { + fn as_str(self) -> &'static str { + match self { + ResolverSwitchReason::StartupPrimary => "startup_primary", + ResolverSwitchReason::ManualOverride => "manual_override", + ResolverSwitchReason::ProbeRecovery => "probe_recovery", + ResolverSwitchReason::TimeoutStreakExceeded => "timeout_streak_exceeded", + ResolverSwitchReason::LossSpike => "loss_spike", + ResolverSwitchReason::LatencyRegression => "latency_regression", + ResolverSwitchReason::PathUnavailable => "path_unavailable", + ResolverSwitchReason::CooldownExpired => "cooldown_expired", + } + } +} + +pub(crate) fn record_resolver_switch( + resolvers: &mut [ResolverState], + from_index: Option, + to_index: usize, + reason: ResolverSwitchReason, +) { + if to_index >= resolvers.len() { + return; + } + + let to_addr = resolvers[to_index].addr; + resolvers[to_index].debug.switch_to_count = + resolvers[to_index].debug.switch_to_count.saturating_add(1); + + let from_addr = if let Some(index) = from_index { + if index < resolvers.len() { + resolvers[index].debug.switch_from_count = + resolvers[index].debug.switch_from_count.saturating_add(1); + Some(resolvers[index].addr) + } else { + None + } + } else { + None + }; + + info!( + "resolver switch: from={} to={} reason={}", + from_addr + .map(|addr| addr.to_string()) + .unwrap_or_else(|| "none".to_string()), + to_addr, + reason.as_str() + ); +} + pub(crate) fn maybe_report_debug( resolver: &mut ResolverState, now: u64, @@ -62,40 +143,44 @@ pub(crate) fn maybe_report_debug( pacing_snapshot: Option, ) { let label = resolver.label(); - let debug = &mut resolver.debug; - if !debug.enabled { + let metrics = &mut resolver.debug; + if !metrics.enabled { return; } - if debug.last_report_at == 0 { - debug.last_report_at = now; + if metrics.last_report_at == 0 { + metrics.last_report_at = now; return; } - let elapsed = now.saturating_sub(debug.last_report_at); + let elapsed = now.saturating_sub(metrics.last_report_at); if elapsed < DEBUG_REPORT_INTERVAL_US { return; } - let dns_delta = debug.dns_responses.saturating_sub(debug.last_report_dns); - let zero_delta = debug.zero_send_loops.saturating_sub(debug.last_report_zero); - let zero_stream_delta = debug + let dns_delta = metrics + .dns_responses + .saturating_sub(metrics.last_report_dns); + let zero_delta = metrics + .zero_send_loops + .saturating_sub(metrics.last_report_zero); + let zero_stream_delta = metrics .zero_send_with_streams - .saturating_sub(debug.last_report_zero_streams); - let data_ready_delta = debug + .saturating_sub(metrics.last_report_zero_streams); + let data_ready_delta = metrics .data_ready_skips - .saturating_sub(debug.last_report_data_ready_skips); - let enq_delta = debug + .saturating_sub(metrics.last_report_data_ready_skips); + let enq_delta = metrics .enqueued_bytes - .saturating_sub(debug.last_report_enqueued); - let send_pkt_delta = debug + .saturating_sub(metrics.last_report_enqueued); + let send_pkt_delta = metrics .send_packets - .saturating_sub(debug.last_report_send_packets); - let send_bytes_delta = debug + .saturating_sub(metrics.last_report_send_packets); + let send_bytes_delta = metrics .send_bytes - .saturating_sub(debug.last_report_send_bytes); - let polls_delta = debug.polls_sent.saturating_sub(debug.last_report_polls); - let enqueue_ms = if debug.last_enqueue_at == 0 { + .saturating_sub(metrics.last_report_send_bytes); + let polls_delta = metrics.polls_sent.saturating_sub(metrics.last_report_polls); + let enqueue_ms = if metrics.last_enqueue_at == 0 { 0 } else { - now.saturating_sub(debug.last_enqueue_at) / 1_000 + now.saturating_sub(metrics.last_enqueue_at) / 1_000 }; let pacing_summary = if let Some(snapshot) = pacing_snapshot { format!( @@ -106,7 +191,7 @@ pub(crate) fn maybe_report_debug( String::new() }; debug!( - "debug: {} dns+={} send_pkts+={} send_bytes+={} polls+={} zero_send+={} zero_send_streams+={} data_ready_skips+={} streams={} enqueued+={} last_enqueue_ms={} pending_polls={} inflight_polls={}{}", + "debug: {} dns+={} send_pkts+={} send_bytes+={} polls+={} zero_send+={} zero_send_streams+={} data_ready_skips+={} streams={} enqueued+={} last_enqueue_ms={} pending_polls={} inflight_polls={} poll_timeouts={} probe_ok={} probe_fail={} path_rtt_us={} path_cwnd={} path_in_transit={} path_pacing_rate={} switches_to={} switches_from={}{}", label, dns_delta, send_pkt_delta, @@ -120,15 +205,62 @@ pub(crate) fn maybe_report_debug( enqueue_ms, pending_polls, inflight_polls, + metrics.inflight_poll_timeouts, + metrics.path_probe_successes, + metrics.path_probe_failures, + metrics.path_rtt_us, + metrics.path_cwnd, + metrics.path_bytes_in_transit, + metrics.path_pacing_rate, + metrics.switch_to_count, + metrics.switch_from_count, pacing_summary ); - debug.last_report_at = now; - debug.last_report_dns = debug.dns_responses; - debug.last_report_zero = debug.zero_send_loops; - debug.last_report_zero_streams = debug.zero_send_with_streams; - debug.last_report_data_ready_skips = debug.data_ready_skips; - debug.last_report_enqueued = debug.enqueued_bytes; - debug.last_report_send_packets = debug.send_packets; - debug.last_report_send_bytes = debug.send_bytes; - debug.last_report_polls = debug.polls_sent; + metrics.last_report_at = now; + metrics.last_report_dns = metrics.dns_responses; + metrics.last_report_zero = metrics.zero_send_loops; + metrics.last_report_zero_streams = metrics.zero_send_with_streams; + metrics.last_report_data_ready_skips = metrics.data_ready_skips; + metrics.last_report_enqueued = metrics.enqueued_bytes; + metrics.last_report_send_packets = metrics.send_packets; + metrics.last_report_send_bytes = metrics.send_bytes; + metrics.last_report_polls = metrics.polls_sent; +} + +#[cfg(test)] +mod tests { + use super::ResolverSwitchReason; + + #[test] + fn resolver_switch_reasons_are_stable_tokens() { + assert_eq!( + ResolverSwitchReason::StartupPrimary.as_str(), + "startup_primary" + ); + assert_eq!( + ResolverSwitchReason::ManualOverride.as_str(), + "manual_override" + ); + assert_eq!( + ResolverSwitchReason::ProbeRecovery.as_str(), + "probe_recovery" + ); + assert_eq!( + ResolverSwitchReason::TimeoutStreakExceeded.as_str(), + "timeout_streak_exceeded" + ); + assert_eq!(ResolverSwitchReason::LossSpike.as_str(), "loss_spike"); + assert_eq!( + ResolverSwitchReason::LatencyRegression.as_str(), + "latency_regression" + ); + assert_eq!( + ResolverSwitchReason::PathUnavailable.as_str(), + "path_unavailable" + ); + assert_eq!( + ResolverSwitchReason::CooldownExpired.as_str(), + "cooldown_expired" + ); + } } diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 5042aa91..3ac2132a 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -86,9 +86,12 @@ pub(crate) fn add_paths( if ret == 0 && path_id >= 0 { resolver.added = true; resolver.path_id = path_id; + resolver.debug.path_probe_successes = + resolver.debug.path_probe_successes.saturating_add(1); info!("Added path {}", resolver.addr); continue; } + resolver.debug.path_probe_failures = resolver.debug.path_probe_failures.saturating_add(1); resolver.probe_attempts = resolver.probe_attempts.saturating_add(1); let delay = path_probe_backoff(resolver.probe_attempts); resolver.next_probe_at = now.saturating_add(delay); diff --git a/crates/slipstream-client/src/dns/poll.rs b/crates/slipstream-client/src/dns/poll.rs index 83e10cd4..234c08d6 100644 --- a/crates/slipstream-client/src/dns/poll.rs +++ b/crates/slipstream-client/src/dns/poll.rs @@ -14,9 +14,9 @@ use slipstream_core::normalize_dual_stack_addr; const AUTHORITATIVE_POLL_TIMEOUT_US: u64 = 5_000_000; -pub(crate) fn expire_inflight_polls(inflight_poll_ids: &mut HashMap, now: u64) { +pub(crate) fn expire_inflight_polls(inflight_poll_ids: &mut HashMap, now: u64) -> usize { if inflight_poll_ids.is_empty() { - return; + return 0; } let expire_before = now.saturating_sub(AUTHORITATIVE_POLL_TIMEOUT_US); let mut expired = Vec::new(); @@ -25,9 +25,11 @@ pub(crate) fn expire_inflight_polls(inflight_poll_ids: &mut HashMap, n expired.push(*id); } } + let expired_count = expired.len(); for id in expired { inflight_poll_ids.remove(&id); } + expired_count } #[allow(clippy::too_many_arguments)] diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index f5c45dce..502658d2 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -1,5 +1,6 @@ use crate::error::ClientError; use crate::pacing::{PacingBudgetSnapshot, PacingPollBudget}; +use slipstream_core::state_machine::ResolverRole; use slipstream_core::{normalize_dual_stack_addr, resolve_host_port}; use slipstream_ffi::{socket_addr_to_storage, ResolverMode, ResolverSpec}; use std::collections::HashMap; @@ -13,6 +14,7 @@ pub(crate) struct ResolverState { pub(crate) storage: libc::sockaddr_storage, pub(crate) local_addr_storage: Option, pub(crate) mode: ResolverMode, + pub(crate) role: ResolverRole, pub(crate) added: bool, pub(crate) path_id: libc::c_int, pub(crate) unique_path_id: Option, @@ -28,8 +30,8 @@ pub(crate) struct ResolverState { impl ResolverState { pub(crate) fn label(&self) -> String { format!( - "path_id={} unique_id={:?} resolver={} mode={:?}", - self.path_id, self.unique_path_id, self.addr, self.mode + "path_id={} unique_id={:?} resolver={} mode={:?} role={:?}", + self.path_id, self.unique_path_id, self.addr, self.mode, self.role ) } } @@ -58,6 +60,11 @@ pub(crate) fn resolve_resolvers( storage: socket_addr_to_storage(addr), local_addr_storage: None, mode: resolver.mode, + role: if is_primary { + ResolverRole::Active + } else { + ResolverRole::Standby + }, added: is_primary, path_id: if is_primary { 0 } else { -1 }, unique_path_id: if is_primary { Some(0) } else { None }, diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 57dacfa6..80ad5e99 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -8,8 +8,8 @@ use self::path::{ use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ add_paths, expire_inflight_polls, handle_dns_response, maybe_report_debug, - refresh_resolver_path, resolve_resolvers, resolver_mode_to_c, send_poll_queries, - sockaddr_storage_to_socket_addr, DnsResponseContext, + record_resolver_switch, refresh_resolver_path, resolve_resolvers, resolver_mode_to_c, + send_poll_queries, sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverSwitchReason, }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -278,6 +278,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if resolvers.is_empty() { return Err(ClientError::new("At least one resolver is required")); } + record_resolver_switch( + &mut resolvers, + None, + 0, + ResolverSwitchReason::StartupPrimary, + ); let mut local_addr_storage = socket_addr_to_storage(udp.local_addr().map_err(map_io)?); @@ -449,7 +455,14 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { for resolver in resolvers.iter_mut() { if resolver.mode == ResolverMode::Authoritative { - expire_inflight_polls(&mut resolver.inflight_poll_ids, current_time); + let expired = + expire_inflight_polls(&mut resolver.inflight_poll_ids, current_time); + if expired > 0 { + resolver.debug.inflight_poll_timeouts = resolver + .debug + .inflight_poll_timeouts + .saturating_add(expired as u64); + } } } @@ -466,6 +479,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let pending_for_sleep = match resolver.mode { ResolverMode::Authoritative => { let quality = fetch_path_quality(cnx, resolver); + resolver.debug.path_rtt_us = quality.rtt; + resolver.debug.path_cwnd = quality.cwin; + resolver.debug.path_bytes_in_transit = quality.bytes_in_transit; + resolver.debug.path_pacing_rate = quality.pacing_rate; let snapshot = resolver .pacing_budget .as_mut() @@ -759,6 +776,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { match resolver.mode { ResolverMode::Authoritative => { let quality = fetch_path_quality(cnx, resolver); + resolver.debug.path_rtt_us = quality.rtt; + resolver.debug.path_cwnd = quality.cwin; + resolver.debug.path_bytes_in_transit = quality.bytes_in_transit; + resolver.debug.path_pacing_rate = quality.pacing_rate; let snapshot = resolver.last_pacing_snapshot; let pacing_target = snapshot .map(|snapshot| snapshot.target_inflight) @@ -879,6 +900,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let pending_for_debug = match resolver.mode { ResolverMode::Authoritative => { let quality = fetch_path_quality(cnx, resolver); + resolver.debug.path_rtt_us = quality.rtt; + resolver.debug.path_cwnd = quality.cwin; + resolver.debug.path_bytes_in_transit = quality.bytes_in_transit; + resolver.debug.path_pacing_rate = quality.pacing_rate; let inflight_packets = inflight_packet_estimate(quality.bytes_in_transit, mtu); resolver From dd659d622ab8ef439ec28ca98b8dba210abd4ec0 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:08:57 +0100 Subject: [PATCH 03/21] refactor(client): introduce resolver manager in parity mode --- crates/slipstream-client/src/dns.rs | 2 +- crates/slipstream-client/src/dns/resolver.rs | 87 +++++++++++++++++++- crates/slipstream-client/src/runtime.rs | 55 +++++++------ 3 files changed, 117 insertions(+), 27 deletions(-) diff --git a/crates/slipstream-client/src/dns.rs b/crates/slipstream-client/src/dns.rs index c7369266..f36c6ef2 100644 --- a/crates/slipstream-client/src/dns.rs +++ b/crates/slipstream-client/src/dns.rs @@ -8,6 +8,6 @@ pub(crate) use debug::{maybe_report_debug, record_resolver_switch, ResolverSwitc pub(crate) use path::{add_paths, refresh_resolver_path, resolver_mode_to_c}; pub(crate) use poll::{expire_inflight_polls, send_poll_queries}; pub(crate) use resolver::{ - reset_resolver_path, resolve_resolvers, sockaddr_storage_to_socket_addr, ResolverState, + reset_resolver_path, sockaddr_storage_to_socket_addr, ResolverManager, ResolverState, }; pub(crate) use response::{handle_dns_response, DnsResponseContext}; diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 502658d2..4a1bfa77 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -9,6 +9,11 @@ use tracing::warn; use super::debug::DebugMetrics; +pub(crate) struct ResolverManager { + resolvers: Vec, + active_index: usize, +} + pub(crate) struct ResolverState { pub(crate) addr: SocketAddr, pub(crate) storage: libc::sockaddr_storage, @@ -36,6 +41,56 @@ impl ResolverState { } } +impl ResolverManager { + pub(crate) fn from_specs( + resolvers: &[ResolverSpec], + mtu: u32, + debug_poll: bool, + ) -> Result { + let resolved = resolve_resolvers(resolvers, mtu, debug_poll)?; + Self::new(resolved) + } + + fn new(mut resolvers: Vec) -> Result { + if resolvers.is_empty() { + return Err(ClientError::new("At least one resolver is required")); + } + + let active_index = resolvers + .iter() + .position(|resolver| resolver.role == ResolverRole::Active) + .unwrap_or(0); + for (index, resolver) in resolvers.iter_mut().enumerate() { + resolver.role = if index == active_index { + ResolverRole::Active + } else { + ResolverRole::Standby + }; + } + + Ok(Self { + resolvers, + active_index, + }) + } + + pub(crate) fn active_index(&self) -> usize { + self.active_index + } + + pub(crate) fn active_mut(&mut self) -> &mut ResolverState { + &mut self.resolvers[self.active_index] + } + + pub(crate) fn as_slice(&self) -> &[ResolverState] { + &self.resolvers + } + + pub(crate) fn as_mut_slice(&mut self) -> &mut [ResolverState] { + &mut self.resolvers + } +} + pub(crate) fn resolve_resolvers( resolvers: &[ResolverSpec], mtu: u32, @@ -107,7 +162,8 @@ pub(crate) fn sockaddr_storage_to_socket_addr( #[cfg(test)] mod tests { - use super::resolve_resolvers; + use super::{resolve_resolvers, ResolverManager}; + use slipstream_core::state_machine::ResolverRole; use slipstream_core::{AddressFamily, HostPort}; use slipstream_ffi::{ResolverMode, ResolverSpec}; @@ -137,4 +193,33 @@ mod tests { Err(err) => assert!(err.to_string().contains("Duplicate resolver address")), } } + + #[test] + fn manager_tracks_single_active_resolver() { + let resolvers = vec![ + ResolverSpec { + resolver: HostPort { + host: "127.0.0.1".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }, + ResolverSpec { + resolver: HostPort { + host: "127.0.0.2".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Authoritative, + }, + ]; + + let manager = ResolverManager::from_specs(&resolvers, 900, false) + .expect("resolver manager should initialize"); + + assert_eq!(manager.active_index(), 0); + assert_eq!(manager.as_slice()[0].role, ResolverRole::Active); + assert_eq!(manager.as_slice()[1].role, ResolverRole::Standby); + } } diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 80ad5e99..067555f9 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -8,8 +8,8 @@ use self::path::{ use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ add_paths, expire_inflight_polls, handle_dns_response, maybe_report_debug, - record_resolver_switch, refresh_resolver_path, resolve_resolvers, resolver_mode_to_c, - send_poll_queries, sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverSwitchReason, + record_resolver_switch, refresh_resolver_path, resolver_mode_to_c, send_poll_queries, + sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverManager, ResolverSwitchReason, }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -274,14 +274,13 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); loop { - let mut resolvers = resolve_resolvers(config.resolvers, mtu, config.debug_poll)?; - if resolvers.is_empty() { - return Err(ClientError::new("At least one resolver is required")); - } + let mut resolver_manager = + ResolverManager::from_specs(config.resolvers, mtu, config.debug_poll)?; + let active_index = resolver_manager.active_index(); record_resolver_switch( - &mut resolvers, + resolver_manager.as_mut_slice(), None, - 0, + active_index, ResolverSwitchReason::StartupPrimary, ); @@ -326,7 +325,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { configure_quic_with_custom(quic, mixed_cc, mtu); // Multipath is only useful with multiple resolvers; leave it off // (picoquic default) for single-resolver to avoid CID exhaustion. - if resolvers.len() > 1 { + if resolver_manager.as_slice().len() > 1 { picoquic_set_default_multipath_option(quic, 1); // Multipath provisions 8 CIDs per path; the default limit of 8 // is too low with 2+ paths. 32 accommodates ~7 paths. @@ -340,12 +339,14 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { slipstream_set_cc_override(override_ptr); } unsafe { - slipstream_set_default_path_mode(resolver_mode_to_c(resolvers[0].mode)); + slipstream_set_default_path_mode(resolver_mode_to_c( + resolver_manager.active_mut().mode, + )); } if let Some(cert) = config.cert { configure_pinned_certificate(quic, cert).map_err(ClientError::new)?; } - let mut server_storage = resolvers[0].storage; + let mut server_storage = resolver_manager.active_mut().storage; // picoquic_create_client_cnx calls picoquic_start_client_cnx internally (see picoquic/quicctx.c). let cnx = unsafe { picoquic_create_client_cnx( @@ -363,7 +364,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { return Err(ClientError::new("Could not create QUIC connection")); } - apply_path_mode(cnx, &mut resolvers[0])?; + apply_path_mode(cnx, resolver_manager.active_mut())?; unsafe { picoquic_set_callback(cnx, Some(client_callback), state_ptr as *mut _); @@ -383,8 +384,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut dns_id = 1u16; let mut recv_buf = vec![0u8; 4096]; let mut send_buf = vec![0u8; PICOQUIC_MAX_PACKET_SIZE]; - let packet_loop_send_max = loop_burst_total(&resolvers, PICOQUIC_PACKET_LOOP_SEND_MAX); - let packet_loop_recv_max = loop_burst_total(&resolvers, PICOQUIC_PACKET_LOOP_RECV_MAX); + let packet_loop_send_max = + loop_burst_total(resolver_manager.as_slice(), PICOQUIC_PACKET_LOOP_SEND_MAX); + let packet_loop_recv_max = + loop_burst_total(resolver_manager.as_slice(), PICOQUIC_PACKET_LOOP_RECV_MAX); let mut zero_send_loops = 0u64; let mut zero_send_with_streams = 0u64; let mut data_ready_skips = 0u64; @@ -444,16 +447,16 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if reconnect_delay != Duration::from_millis(RECONNECT_SLEEP_MIN_MS) { reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); } - add_paths(cnx, &mut resolvers)?; - for resolver in resolvers.iter_mut() { + add_paths(cnx, resolver_manager.as_mut_slice())?; + for resolver in resolver_manager.as_mut_slice().iter_mut() { if resolver.added { apply_path_mode(cnx, resolver)?; } } } - drain_path_events(cnx, &mut resolvers, state_ptr); + drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().iter_mut() { if resolver.mode == ResolverMode::Authoritative { let expired = expire_inflight_polls(&mut resolver.inflight_poll_ids, current_time); @@ -472,7 +475,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let delay_us = if delay_us < 0 { 0 } else { delay_us as u64 }; let streams_len_for_sleep = unsafe { (*state_ptr).streams_len() }; let mut has_work = streams_len_for_sleep > 0; - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().iter_mut() { if !refresh_resolver_path(cnx, resolver) { continue; } @@ -541,7 +544,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut response_ctx = DnsResponseContext { quic, local_addr_storage: &local_addr_storage, - resolvers: &mut resolvers, + resolvers: resolver_manager.as_mut_slice(), }; handle_dns_response(&recv_buf[..size], peer, &mut response_ctx)?; for _ in 1..packet_loop_recv_max { @@ -588,7 +591,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let _ = drain_disconnected_commands(&mut command_rx); } drain_stream_data(cnx, state_ptr); - drain_path_events(cnx, &mut resolvers, state_ptr); + drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); let reaped_half_closed = unsafe { let now = picoquic_current_time(); (*state_ptr).reap_stale_half_closed_streams(now, HALF_CLOSE_IDLE_TIMEOUT_US) @@ -682,7 +685,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { // congestion window is full (not just flow control). // The server can only send ACKs inside DNS responses, // so we must keep querying to unblock the CWND. - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().iter_mut() { if resolver.mode == ResolverMode::Recursive && resolver.added { resolver.pending_polls = resolver.pending_polls.max(1); } @@ -697,7 +700,9 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { sent_quic_data = true; if let Ok(dest) = sockaddr_storage_to_socket_addr(&addr_to) { let dest = normalize_dual_stack_addr(dest); - if let Some(resolver) = find_resolver_by_addr_mut(&mut resolvers, dest) { + if let Some(resolver) = + find_resolver_by_addr_mut(resolver_manager.as_mut_slice(), dest) + { resolver.local_addr_storage = Some(unsafe { std::ptr::read(&addr_from) }); resolver.debug.send_packets = resolver.debug.send_packets.saturating_add(1); resolver.debug.send_bytes = @@ -769,7 +774,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } watchdog.set_phase(PHASE_POLL_QUERIES); - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().iter_mut() { if !refresh_resolver_path(cnx, resolver) { continue; } @@ -887,7 +892,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let report_time = unsafe { picoquic_current_time() }; let (enqueued_bytes, last_enqueue_at) = unsafe { (*state_ptr).debug_snapshot() }; let streams_len = unsafe { (*state_ptr).streams_len() }; - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().iter_mut() { resolver.debug.enqueued_bytes = enqueued_bytes; resolver.debug.last_enqueue_at = last_enqueue_at; resolver.debug.zero_send_loops = zero_send_loops; From a37e01c9c36601ae1ca5cdb82c69531f194d2ba1 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:11:27 +0100 Subject: [PATCH 04/21] feat(client): add adaptive single-active resolver selection --- crates/slipstream-client/src/dns/resolver.rs | 173 +++++++++++++++++++ crates/slipstream-client/src/runtime.rs | 26 ++- 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 4a1bfa77..064173cf 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -8,10 +8,22 @@ use std::net::SocketAddr; use tracing::warn; use super::debug::DebugMetrics; +use super::debug::ResolverSwitchReason; + +const SELECTOR_COOLDOWN_US: u64 = 10_000_000; +const SELECTOR_PROMOTION_THRESHOLD_PERCENT: u64 = 15; +const SELECTOR_REQUIRED_CONSECUTIVE_WINS: u8 = 3; +const SCORE_UNAVAILABLE_PENALTY: u64 = 2_000_000; +const SCORE_TIMEOUT_PENALTY: u64 = 50_000; +const SCORE_PROBE_FAILURE_PENALTY: u64 = 20_000; +const SCORE_STICKINESS_BIAS: u64 = 10_000; pub(crate) struct ResolverManager { resolvers: Vec, active_index: usize, + last_switch_at: u64, + candidate_index: Option, + candidate_wins: u8, } pub(crate) struct ResolverState { @@ -39,6 +51,10 @@ impl ResolverState { self.path_id, self.unique_path_id, self.addr, self.mode, self.role ) } + + pub(crate) fn is_active(&self) -> bool { + self.role == ResolverRole::Active + } } impl ResolverManager { @@ -71,6 +87,9 @@ impl ResolverManager { Ok(Self { resolvers, active_index, + last_switch_at: 0, + candidate_index: None, + candidate_wins: 0, }) } @@ -82,6 +101,10 @@ impl ResolverManager { &mut self.resolvers[self.active_index] } + pub(crate) fn active(&self) -> &ResolverState { + &self.resolvers[self.active_index] + } + pub(crate) fn as_slice(&self) -> &[ResolverState] { &self.resolvers } @@ -89,6 +112,117 @@ impl ResolverManager { pub(crate) fn as_mut_slice(&mut self) -> &mut [ResolverState] { &mut self.resolvers } + + pub(crate) fn maybe_select_active( + &mut self, + now_us: u64, + ) -> Option<(usize, usize, ResolverSwitchReason)> { + if self.resolvers.len() <= 1 { + return None; + } + + let from = self.active_index; + if !self.resolvers[from].added { + if let Some(to) = self.best_available_index() { + if to != from { + self.set_active(to, now_us); + return Some((from, to, ResolverSwitchReason::PathUnavailable)); + } + } + return None; + } + + let to = self.best_available_index()?; + if to == from { + self.candidate_index = None; + self.candidate_wins = 0; + return None; + } + + let current_score = self.score(from); + let candidate_score = self.score(to); + let required_improvement = + current_score.saturating_mul(SELECTOR_PROMOTION_THRESHOLD_PERCENT) / 100; + let actual_improvement = current_score.saturating_sub(candidate_score); + if actual_improvement < required_improvement { + self.candidate_index = None; + self.candidate_wins = 0; + return None; + } + + if self.candidate_index == Some(to) { + self.candidate_wins = self.candidate_wins.saturating_add(1); + } else { + self.candidate_index = Some(to); + self.candidate_wins = 1; + } + if self.candidate_wins < SELECTOR_REQUIRED_CONSECUTIVE_WINS { + return None; + } + if self.last_switch_at > 0 + && now_us.saturating_sub(self.last_switch_at) < SELECTOR_COOLDOWN_US + { + return None; + } + + self.set_active(to, now_us); + Some((from, to, ResolverSwitchReason::LatencyRegression)) + } + + fn set_active(&mut self, active_index: usize, now_us: u64) { + self.active_index = active_index; + for (index, resolver) in self.resolvers.iter_mut().enumerate() { + resolver.role = if index == active_index { + ResolverRole::Active + } else { + ResolverRole::Standby + }; + } + self.last_switch_at = now_us; + self.candidate_index = None; + self.candidate_wins = 0; + } + + fn best_available_index(&self) -> Option { + let mut best_index = None; + let mut best_score = u64::MAX; + for (index, resolver) in self.resolvers.iter().enumerate() { + if !resolver.added { + continue; + } + let score = self.score(index); + if score < best_score { + best_score = score; + best_index = Some(index); + } + } + best_index + } + + fn score(&self, index: usize) -> u64 { + let resolver = &self.resolvers[index]; + let mut score = resolver.debug.path_rtt_us.max(100_000); + score = score.saturating_add( + resolver + .debug + .inflight_poll_timeouts + .saturating_mul(SCORE_TIMEOUT_PENALTY), + ); + score = score.saturating_add( + resolver + .debug + .path_probe_failures + .saturating_mul(SCORE_PROBE_FAILURE_PENALTY), + ); + score = score.saturating_add(resolver.debug.path_bytes_in_transit / 8); + if !resolver.added { + score = score.saturating_add(SCORE_UNAVAILABLE_PENALTY); + } + if resolver.role == ResolverRole::Active { + score = score.saturating_sub(SCORE_STICKINESS_BIAS); + } + score + } } pub(crate) fn resolve_resolvers( @@ -222,4 +356,43 @@ mod tests { assert_eq!(manager.as_slice()[0].role, ResolverRole::Active); assert_eq!(manager.as_slice()[1].role, ResolverRole::Standby); } + + #[test] + fn manager_switches_active_when_candidate_consistently_better() { + let resolvers = vec![ + ResolverSpec { + resolver: HostPort { + host: "127.0.0.1".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }, + ResolverSpec { + resolver: HostPort { + host: "127.0.0.2".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }, + ]; + + let mut manager = ResolverManager::from_specs(&resolvers, 900, false) + .expect("resolver manager should initialize"); + manager.as_mut_slice()[0].added = true; + manager.as_mut_slice()[1].added = true; + manager.as_mut_slice()[0].debug.path_rtt_us = 300_000; + manager.as_mut_slice()[1].debug.path_rtt_us = 50_000; + + assert!(manager.maybe_select_active(1_000_000).is_none()); + assert!(manager.maybe_select_active(2_000_000).is_none()); + let switch = manager + .maybe_select_active(11_000_000) + .expect("third consecutive win should switch"); + assert_eq!(switch.0, 0); + assert_eq!(switch.1, 1); + assert_eq!(manager.active_index(), 1); + assert_eq!(manager.active().role, ResolverRole::Active); + } } diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 067555f9..c5b934b3 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -479,6 +479,9 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if !refresh_resolver_path(cnx, resolver) { continue; } + if !resolver.is_active() { + continue; + } let pending_for_sleep = match resolver.mode { ResolverMode::Authoritative => { let quality = fetch_path_quality(cnx, resolver); @@ -686,7 +689,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { // The server can only send ACKs inside DNS responses, // so we must keep querying to unblock the CWND. for resolver in resolver_manager.as_mut_slice().iter_mut() { - if resolver.mode == ResolverMode::Recursive && resolver.added { + if resolver.is_active() + && resolver.mode == ResolverMode::Recursive + && resolver.added + { resolver.pending_polls = resolver.pending_polls.max(1); } } @@ -737,6 +743,21 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } + if let Some((from_index, to_index, reason)) = + resolver_manager.maybe_select_active(current_time) + { + record_resolver_switch( + resolver_manager.as_mut_slice(), + Some(from_index), + to_index, + reason, + ); + unsafe { + let mode = resolver_manager.active().mode; + slipstream_set_default_path_mode(resolver_mode_to_c(mode)); + } + } + let has_ready_stream = unsafe { slipstream_has_ready_stream(cnx) != 0 }; let flow_blocked = unsafe { slipstream_is_flow_blocked(cnx) != 0 }; let streams_len = unsafe { (*state_ptr).streams_len() }; @@ -775,6 +796,9 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } watchdog.set_phase(PHASE_POLL_QUERIES); for resolver in resolver_manager.as_mut_slice().iter_mut() { + if !resolver.is_active() { + continue; + } if !refresh_resolver_path(cnx, resolver) { continue; } From d3b1c1411ed63570b1559f9368b43de9899b2fb9 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:20:12 +0100 Subject: [PATCH 05/21] fix(client): satisfy dead-code lint for switch reasons --- crates/slipstream-client/src/dns.rs | 5 ++++- crates/slipstream-client/src/dns/debug.rs | 14 ++++++++++++++ crates/slipstream-client/src/runtime.rs | 6 ++++-- 3 files changed, 22 insertions(+), 3 deletions(-) diff --git a/crates/slipstream-client/src/dns.rs b/crates/slipstream-client/src/dns.rs index f36c6ef2..4587cf52 100644 --- a/crates/slipstream-client/src/dns.rs +++ b/crates/slipstream-client/src/dns.rs @@ -4,7 +4,10 @@ mod poll; mod resolver; mod response; -pub(crate) use debug::{maybe_report_debug, record_resolver_switch, ResolverSwitchReason}; +pub(crate) use debug::{ + maybe_report_debug, record_resolver_switch, resolver_switch_reason_catalog, + ResolverSwitchReason, +}; pub(crate) use path::{add_paths, refresh_resolver_path, resolver_mode_to_c}; pub(crate) use poll::{expire_inflight_polls, send_poll_queries}; pub(crate) use resolver::{ diff --git a/crates/slipstream-client/src/dns/debug.rs b/crates/slipstream-client/src/dns/debug.rs index 417606a5..c4625673 100644 --- a/crates/slipstream-client/src/dns/debug.rs +++ b/crates/slipstream-client/src/dns/debug.rs @@ -83,6 +83,20 @@ pub(crate) enum ResolverSwitchReason { CooldownExpired, } +pub(crate) fn resolver_switch_reason_catalog() -> &'static [ResolverSwitchReason] { + static REASONS: [ResolverSwitchReason; 8] = [ + ResolverSwitchReason::StartupPrimary, + ResolverSwitchReason::ManualOverride, + ResolverSwitchReason::ProbeRecovery, + ResolverSwitchReason::TimeoutStreakExceeded, + ResolverSwitchReason::LossSpike, + ResolverSwitchReason::LatencyRegression, + ResolverSwitchReason::PathUnavailable, + ResolverSwitchReason::CooldownExpired, + ]; + &REASONS +} + impl ResolverSwitchReason { fn as_str(self) -> &'static str { match self { diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index c5b934b3..c0f7e7a7 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -8,8 +8,9 @@ use self::path::{ use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ add_paths, expire_inflight_polls, handle_dns_response, maybe_report_debug, - record_resolver_switch, refresh_resolver_path, resolver_mode_to_c, send_poll_queries, - sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverManager, ResolverSwitchReason, + record_resolver_switch, refresh_resolver_path, resolver_mode_to_c, + resolver_switch_reason_catalog, send_poll_queries, sockaddr_storage_to_socket_addr, + DnsResponseContext, ResolverManager, ResolverSwitchReason, }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -210,6 +211,7 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver } pub async fn run_client(config: &ClientConfig<'_>) -> Result { + let _ = resolver_switch_reason_catalog(); let domain_len = config.domain.len(); let mtu = compute_mtu(domain_len)?; let udp = bind_udp_socket().await?; From d575d5f5b69dac0f01a0daa381bb268956747c9c Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 09:47:54 +0100 Subject: [PATCH 06/21] fix(client): preserve active resolver across reconnects --- crates/slipstream-client/src/dns/resolver.rs | 62 ++++++++++++++++++++ crates/slipstream-client/src/runtime.rs | 3 + 2 files changed, 65 insertions(+) diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 064173cf..9a7a1fdf 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -97,6 +97,34 @@ impl ResolverManager { self.active_index } + pub(crate) fn set_startup_active_index(&mut self, index: usize) { + if index >= self.resolvers.len() || index == self.active_index { + return; + } + + self.active_index = index; + for (resolver_index, resolver) in self.resolvers.iter_mut().enumerate() { + resolver.role = if resolver_index == index { + ResolverRole::Active + } else { + ResolverRole::Standby + }; + resolver.added = resolver_index == index; + resolver.path_id = if resolver_index == index { 0 } else { -1 }; + resolver.unique_path_id = if resolver_index == index { + Some(0) + } else { + None + }; + resolver.local_addr_storage = None; + resolver.pending_polls = 0; + resolver.inflight_poll_ids.clear(); + resolver.last_pacing_snapshot = None; + resolver.probe_attempts = 0; + resolver.next_probe_at = 0; + } + } + pub(crate) fn active_mut(&mut self) -> &mut ResolverState { &mut self.resolvers[self.active_index] } @@ -395,4 +423,38 @@ mod tests { assert_eq!(manager.active_index(), 1); assert_eq!(manager.active().role, ResolverRole::Active); } + + #[test] + fn manager_can_set_startup_active_index() { + let resolvers = vec![ + ResolverSpec { + resolver: HostPort { + host: "127.0.0.1".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }, + ResolverSpec { + resolver: HostPort { + host: "127.0.0.2".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }, + ]; + + let mut manager = ResolverManager::from_specs(&resolvers, 900, false) + .expect("resolver manager should initialize"); + manager.set_startup_active_index(1); + + assert_eq!(manager.active_index(), 1); + assert!(manager.as_slice()[1].added); + assert_eq!(manager.as_slice()[1].path_id, 0); + assert_eq!(manager.as_slice()[1].unique_path_id, Some(0)); + assert!(!manager.as_slice()[0].added); + assert_eq!(manager.as_slice()[0].path_id, -1); + assert_eq!(manager.as_slice()[0].unique_path_id, None); + } } diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index c0f7e7a7..49b55299 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -274,10 +274,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let _state = state; let mut reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); + let mut preferred_startup_resolver_index = 0usize; loop { let mut resolver_manager = ResolverManager::from_specs(config.resolvers, mtu, config.debug_poll)?; + resolver_manager.set_startup_active_index(preferred_startup_resolver_index); let active_index = resolver_manager.active_index(); record_resolver_switch( resolver_manager.as_mut_slice(), @@ -748,6 +750,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if let Some((from_index, to_index, reason)) = resolver_manager.maybe_select_active(current_time) { + preferred_startup_resolver_index = to_index; record_resolver_switch( resolver_manager.as_mut_slice(), Some(from_index), From 4f0dbe90bab79e40b1366e2cbc987201389338f3 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 10:30:13 +0100 Subject: [PATCH 07/21] fix(client): reconnect early on active path loss under load --- crates/slipstream-client/src/runtime.rs | 78 +++++++++++++++----- crates/slipstream-client/src/runtime/path.rs | 12 ++- 2 files changed, 70 insertions(+), 20 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 49b55299..0de21b54 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -72,6 +72,7 @@ const ACCEPTOR_SATURATED_TIMEOUT_US: u64 = 30_000_000; const HEALTH_LOG_INTERVAL_US: u64 = 300_000_000; const WATCHDOG_STALE_SECS: u64 = 15; const WATCHDOG_CHECK_INTERVAL: Duration = Duration::from_secs(3); +const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32; /// Watchdog that runs on a separate OS thread (not tokio) to detect when the /// single-threaded tokio runtime freezes (e.g. a picoquic C FFI call hangs). @@ -210,6 +211,27 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver dropped } +fn maybe_switch_active_resolver( + resolver_manager: &mut ResolverManager, + current_time: u64, + preferred_startup_resolver_index: &mut usize, +) { + if let Some((from_index, to_index, reason)) = resolver_manager.maybe_select_active(current_time) + { + *preferred_startup_resolver_index = to_index; + record_resolver_switch( + resolver_manager.as_mut_slice(), + Some(from_index), + to_index, + reason, + ); + unsafe { + let mode = resolver_manager.active().mode; + slipstream_set_default_path_mode(resolver_mode_to_c(mode)); + } + } +} + pub async fn run_client(config: &ClientConfig<'_>) -> Result { let _ = resolver_switch_reason_catalog(); let domain_len = config.domain.len(); @@ -458,7 +480,23 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } } - drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); + let active_path_deleted = + drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); + if active_path_deleted { + maybe_switch_active_resolver( + &mut resolver_manager, + current_time, + &mut preferred_startup_resolver_index, + ); + let streams_len = unsafe { (*state_ptr).streams_len() }; + if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS { + warn!( + "active resolver path deleted with {} streams; reconnecting to limit reset storm", + streams_len + ); + break; + } + } for resolver in resolver_manager.as_mut_slice().iter_mut() { if resolver.mode == ResolverMode::Authoritative { @@ -598,7 +636,23 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let _ = drain_disconnected_commands(&mut command_rx); } drain_stream_data(cnx, state_ptr); - drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); + let active_path_deleted = + drain_path_events(cnx, resolver_manager.as_mut_slice(), state_ptr); + if active_path_deleted { + maybe_switch_active_resolver( + &mut resolver_manager, + current_time, + &mut preferred_startup_resolver_index, + ); + let streams_len = unsafe { (*state_ptr).streams_len() }; + if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS { + warn!( + "active resolver path deleted with {} streams; reconnecting to limit reset storm", + streams_len + ); + break; + } + } let reaped_half_closed = unsafe { let now = picoquic_current_time(); (*state_ptr).reap_stale_half_closed_streams(now, HALF_CLOSE_IDLE_TIMEOUT_US) @@ -747,21 +801,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } - if let Some((from_index, to_index, reason)) = - resolver_manager.maybe_select_active(current_time) - { - preferred_startup_resolver_index = to_index; - record_resolver_switch( - resolver_manager.as_mut_slice(), - Some(from_index), - to_index, - reason, - ); - unsafe { - let mode = resolver_manager.active().mode; - slipstream_set_default_path_mode(resolver_mode_to_c(mode)); - } - } + maybe_switch_active_resolver( + &mut resolver_manager, + current_time, + &mut preferred_startup_resolver_index, + ); let has_ready_stream = unsafe { slipstream_has_ready_stream(cnx) != 0 }; let flow_blocked = unsafe { slipstream_is_flow_blocked(cnx) != 0 }; diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index 966a4e75..e4355d92 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -51,14 +51,15 @@ pub(crate) fn drain_path_events( cnx: *mut picoquic_cnx_t, resolvers: &mut [ResolverState], state_ptr: *mut ClientState, -) { +) -> bool { if state_ptr.is_null() { - return; + return false; } let events = unsafe { (*state_ptr).take_path_events() }; if events.is_empty() { - return; + return false; } + let mut active_path_deleted = false; for event in events { match event { PathEvent::Available(unique_path_id) => { @@ -78,11 +79,16 @@ pub(crate) fn drain_path_events( } PathEvent::Deleted(unique_path_id) => { if let Some(resolver) = find_resolver_by_unique_id_mut(resolvers, unique_path_id) { + if resolver.is_active() { + active_path_deleted = true; + } reset_resolver_path(resolver); } } } } + + active_path_deleted } fn path_peer_addr(cnx: *mut picoquic_cnx_t, unique_path_id: u64) -> Option { From 8c5c0952f8f43fb0dec92fffcb5471e432f161e2 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 10:35:09 +0100 Subject: [PATCH 08/21] fix(client): avoid unnecessary reconnect when standby path is ready --- crates/slipstream-client/src/runtime.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 0de21b54..4134c969 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -489,10 +489,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { &mut preferred_startup_resolver_index, ); let streams_len = unsafe { (*state_ptr).streams_len() }; - if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS { + let active_path_ready = resolver_manager.active().added; + if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS && !active_path_ready { warn!( - "active resolver path deleted with {} streams; reconnecting to limit reset storm", - streams_len + "active resolver path deleted with {} streams and no standby path ready; reconnecting to limit reset storm", + streams_len, ); break; } @@ -645,10 +646,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { &mut preferred_startup_resolver_index, ); let streams_len = unsafe { (*state_ptr).streams_len() }; - if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS { + let active_path_ready = resolver_manager.active().added; + if streams_len >= ACTIVE_PATH_LOSS_RECONNECT_STREAMS && !active_path_ready { warn!( - "active resolver path deleted with {} streams; reconnecting to limit reset storm", - streams_len + "active resolver path deleted with {} streams and no standby path ready; reconnecting to limit reset storm", + streams_len, ); break; } From d93cb24e67fe367d7adc942e0435eebca9dc75be Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 00:33:26 +0100 Subject: [PATCH 09/21] fix(client): rotate startup resolver after handshake stall --- crates/slipstream-client/src/runtime.rs | 81 ++++++++++++++++++++ crates/slipstream-client/src/runtime/path.rs | 42 +++++++++- 2 files changed, 119 insertions(+), 4 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 4134c969..b9e27b13 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -1,16 +1,35 @@ mod path; mod setup; +<<<<<<< HEAD use self::path::{ apply_path_mode, drain_path_events, fetch_path_quality, find_resolver_by_addr_mut, loop_burst_total, path_poll_burst_max, +======= +use self::actions::{poll_authoritative_resolver, poll_recursive_resolver, PollDispatch}; +use self::deadlock::AcceptorSaturationTracker; +use self::decision::{reconnect_due_to_acceptor_deadlock, reconnect_due_to_active_path_loss}; +use self::health::{ + compute_last_enqueue_ms, should_log_flow_blocked, should_log_health, + should_reconnect_for_handshake_stall, should_reconnect_for_resolver_stall, +}; +use self::loop_policy::compute_loop_sleep_policy; +use self::path::{ + apply_path_mode, drain_path_events, fetch_and_record_path_quality, find_resolver_by_addr_mut, + loop_burst_total, maybe_switch_active_resolver, +>>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) }; use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ add_paths, expire_inflight_polls, handle_dns_response, maybe_report_debug, record_resolver_switch, refresh_resolver_path, resolver_mode_to_c, +<<<<<<< HEAD resolver_switch_reason_catalog, send_poll_queries, sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverManager, ResolverSwitchReason, +======= + resolver_switch_reason_catalog, sockaddr_storage_to_socket_addr, DnsResponseContext, + ResolverManager, ResolverSwitchReason, +>>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -59,6 +78,10 @@ const MIN_POLL_INTERVAL_US: u64 = 100; /// connection is active. This catches cases where the recursive resolver /// silently stops forwarding queries (rate-limit, anti-tunnel heuristic, etc.). const RESOLVER_STALL_TIMEOUT_US: u64 = 60_000_000; +/// Force reconnect if the connection never reaches ready state within this +/// startup window. With multiple recursive resolvers, rotate the startup +/// resolver index on each stall to avoid sticking on a blocked resolver. +const HANDSHAKE_STALL_TIMEOUT_US: u64 = 30_000_000; /// If a stream is half-closed by the remote side and local upload stays open /// without forward progress for too long, abort the local reader to avoid /// zombie streams exhausting MAX_STREAMS credit. @@ -211,6 +234,7 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver dropped } +<<<<<<< HEAD fn maybe_switch_active_resolver( resolver_manager: &mut ResolverManager, current_time: u64, @@ -229,6 +253,21 @@ fn maybe_switch_active_resolver( let mode = resolver_manager.active().mode; slipstream_set_default_path_mode(resolver_mode_to_c(mode)); } +======= +fn drain_commands_by_connection_state( + cnx: *mut picoquic_cnx_t, + state_ptr: *mut ClientState, + command_rx: &mut mpsc::UnboundedReceiver, +) { + // Only process application commands after the QUIC handshake + // completes. During reconnect the acceptor may queue NewStream + // commands while the connection is still in initial state; + // processing them before ready triggers picoquic errors (0xc). + if unsafe { (*state_ptr).is_ready() } { + drain_commands(cnx, state_ptr, command_rx); + } else { + let _ = drain_disconnected_commands(command_rx); +>>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) } } @@ -313,6 +352,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut local_addr_storage = socket_addr_to_storage(udp.local_addr().map_err(map_io)?); let current_time = unsafe { picoquic_current_time() }; + let connect_started_at = current_time; let quic = unsafe { picoquic_create( 8, @@ -463,6 +503,39 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } let ready = unsafe { (*state_ptr).is_ready() }; + if let Some(stall_us) = should_reconnect_for_handshake_stall( + ready, + current_time, + connect_started_at, + HANDSHAKE_STALL_TIMEOUT_US, + ) { + let resolver_count = resolver_manager.as_slice().len(); + if resolver_count > 1 { + let from_index = resolver_manager.active_index(); + let to_index = (from_index + 1) % resolver_count; + let from_addr = resolver_manager.as_slice()[from_index].addr; + let to_addr = resolver_manager.as_slice()[to_index].addr; + record_resolver_switch( + resolver_manager.as_mut_slice(), + Some(from_index), + to_index, + ResolverSwitchReason::TimeoutStreakExceeded, + ); + preferred_startup_resolver_index = to_index; + warn!( + "handshake stall for {:.1}s on startup resolver {}; rotating startup resolver to {} and reconnecting", + stall_us as f64 / 1_000_000.0, + from_addr, + to_addr, + ); + } else { + warn!( + "handshake stall for {:.1}s on single resolver; reconnecting", + stall_us as f64 / 1_000_000.0 + ); + } + break; + } if ready { if last_recv_at == 0 { last_recv_at = current_time; @@ -644,6 +717,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { &mut resolver_manager, current_time, &mut preferred_startup_resolver_index, +<<<<<<< HEAD ); let streams_len = unsafe { (*state_ptr).streams_len() }; let active_path_ready = resolver_manager.active().added; @@ -654,6 +728,13 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { ); break; } +======= + state_ptr, + ACTIVE_PATH_LOSS_RECONNECT_STREAMS, + ) + { + break; +>>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) } let reaped_half_closed = unsafe { let now = picoquic_current_time(); diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index e4355d92..7128188b 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -1,14 +1,15 @@ use crate::dns::{ - refresh_resolver_path, reset_resolver_path, resolver_mode_to_c, - sockaddr_storage_to_socket_addr, ResolverState, + record_resolver_switch, refresh_resolver_path, reset_resolver_path, resolver_mode_to_c, + sockaddr_storage_to_socket_addr, ResolverManager, ResolverState, }; use crate::error::ClientError; use crate::streams::{ClientState, PathEvent}; use slipstream_core::normalize_dual_stack_addr; use slipstream_ffi::picoquic::{ picoquic_cnx_t, picoquic_get_default_path_quality, picoquic_get_path_addr, - picoquic_get_path_quality, slipstream_get_path_id_from_unique, slipstream_set_path_ack_delay, - slipstream_set_path_mode, PICOQUIC_PACKET_LOOP_SEND_MAX, + picoquic_get_path_quality, slipstream_get_path_id_from_unique, + slipstream_set_default_path_mode, slipstream_set_path_ack_delay, slipstream_set_path_mode, + PICOQUIC_PACKET_LOOP_SEND_MAX, }; use slipstream_ffi::ResolverMode; use std::net::SocketAddr; @@ -47,6 +48,39 @@ pub(crate) fn fetch_path_quality( quality } +pub(crate) fn fetch_and_record_path_quality( + cnx: *mut picoquic_cnx_t, + resolver: &mut ResolverState, +) -> slipstream_ffi::picoquic::picoquic_path_quality_t { + let quality = fetch_path_quality(cnx, resolver); + resolver.debug.path_rtt_us = quality.rtt; + resolver.debug.path_cwnd = quality.cwin; + resolver.debug.path_bytes_in_transit = quality.bytes_in_transit; + resolver.debug.path_pacing_rate = quality.pacing_rate; + quality +} + +pub(crate) fn maybe_switch_active_resolver( + resolver_manager: &mut ResolverManager, + current_time: u64, + preferred_startup_resolver_index: &mut usize, +) { + if let Some((from_index, to_index, reason)) = resolver_manager.maybe_select_active(current_time) + { + *preferred_startup_resolver_index = to_index; + record_resolver_switch( + resolver_manager.as_mut_slice(), + Some(from_index), + to_index, + reason, + ); + unsafe { + let mode = resolver_manager.active().mode; + slipstream_set_default_path_mode(resolver_mode_to_c(mode)); + } + } +} + pub(crate) fn drain_path_events( cnx: *mut picoquic_cnx_t, resolvers: &mut [ResolverState], From e0855ddcb950b760868ecb651cf0c842f1b5d343 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 09:51:25 +0100 Subject: [PATCH 10/21] fix(client): add explicit resolver reason for handshake stall --- crates/slipstream-client/src/dns/debug.rs | 9 ++++++++- crates/slipstream-client/src/runtime.rs | 2 +- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/slipstream-client/src/dns/debug.rs b/crates/slipstream-client/src/dns/debug.rs index c4625673..03e9f2f0 100644 --- a/crates/slipstream-client/src/dns/debug.rs +++ b/crates/slipstream-client/src/dns/debug.rs @@ -77,6 +77,7 @@ pub(crate) enum ResolverSwitchReason { ManualOverride, ProbeRecovery, TimeoutStreakExceeded, + HandshakeStall, LossSpike, LatencyRegression, PathUnavailable, @@ -84,11 +85,12 @@ pub(crate) enum ResolverSwitchReason { } pub(crate) fn resolver_switch_reason_catalog() -> &'static [ResolverSwitchReason] { - static REASONS: [ResolverSwitchReason; 8] = [ + static REASONS: [ResolverSwitchReason; 9] = [ ResolverSwitchReason::StartupPrimary, ResolverSwitchReason::ManualOverride, ResolverSwitchReason::ProbeRecovery, ResolverSwitchReason::TimeoutStreakExceeded, + ResolverSwitchReason::HandshakeStall, ResolverSwitchReason::LossSpike, ResolverSwitchReason::LatencyRegression, ResolverSwitchReason::PathUnavailable, @@ -104,6 +106,7 @@ impl ResolverSwitchReason { ResolverSwitchReason::ManualOverride => "manual_override", ResolverSwitchReason::ProbeRecovery => "probe_recovery", ResolverSwitchReason::TimeoutStreakExceeded => "timeout_streak_exceeded", + ResolverSwitchReason::HandshakeStall => "handshake_stall", ResolverSwitchReason::LossSpike => "loss_spike", ResolverSwitchReason::LatencyRegression => "latency_regression", ResolverSwitchReason::PathUnavailable => "path_unavailable", @@ -263,6 +266,10 @@ mod tests { ResolverSwitchReason::TimeoutStreakExceeded.as_str(), "timeout_streak_exceeded" ); + assert_eq!( + ResolverSwitchReason::HandshakeStall.as_str(), + "handshake_stall" + ); assert_eq!(ResolverSwitchReason::LossSpike.as_str(), "loss_spike"); assert_eq!( ResolverSwitchReason::LatencyRegression.as_str(), diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index b9e27b13..80bb1fd0 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -519,7 +519,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { resolver_manager.as_mut_slice(), Some(from_index), to_index, - ResolverSwitchReason::TimeoutStreakExceeded, + ResolverSwitchReason::HandshakeStall, ); preferred_startup_resolver_index = to_index; warn!( From fe4a73f820117ec7ac6148710a1ad41cb86d2b6c Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 10:16:43 +0100 Subject: [PATCH 11/21] fix(client): avoid synthetic startup path unique id --- crates/slipstream-client/src/dns/resolver.rs | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 9a7a1fdf..c74f6f2a 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -111,11 +111,7 @@ impl ResolverManager { }; resolver.added = resolver_index == index; resolver.path_id = if resolver_index == index { 0 } else { -1 }; - resolver.unique_path_id = if resolver_index == index { - Some(0) - } else { - None - }; + resolver.unique_path_id = None; resolver.local_addr_storage = None; resolver.pending_polls = 0; resolver.inflight_poll_ids.clear(); @@ -284,7 +280,7 @@ pub(crate) fn resolve_resolvers( }, added: is_primary, path_id: if is_primary { 0 } else { -1 }, - unique_path_id: if is_primary { Some(0) } else { None }, + unique_path_id: None, probe_attempts: 0, next_probe_at: 0, pending_polls: 0, @@ -452,7 +448,7 @@ mod tests { assert_eq!(manager.active_index(), 1); assert!(manager.as_slice()[1].added); assert_eq!(manager.as_slice()[1].path_id, 0); - assert_eq!(manager.as_slice()[1].unique_path_id, Some(0)); + assert_eq!(manager.as_slice()[1].unique_path_id, None); assert!(!manager.as_slice()[0].added); assert_eq!(manager.as_slice()[0].path_id, -1); assert_eq!(manager.as_slice()[0].unique_path_id, None); From 92a1e9bd8d3d0f565ada8028a401926974842bea Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 10:38:59 +0100 Subject: [PATCH 12/21] fix(client): require confirmed active path delete before failover --- crates/slipstream-client/src/dns/path.rs | 4 ++ crates/slipstream-client/src/dns/resolver.rs | 8 +++ crates/slipstream-client/src/runtime/path.rs | 68 +++++++++++++++++++- 3 files changed, 79 insertions(+), 1 deletion(-) diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 3ac2132a..d2425277 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -23,6 +23,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } + resolver.active_delete_suspect_count = 0; + resolver.active_delete_first_at = 0; return true; } resolver.unique_path_id = None; @@ -40,6 +42,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } + resolver.active_delete_suspect_count = 0; + resolver.active_delete_first_at = 0; true } diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index c74f6f2a..8e65d8a5 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -42,6 +42,8 @@ pub(crate) struct ResolverState { pub(crate) pacing_budget: Option, pub(crate) last_pacing_snapshot: Option, pub(crate) debug: DebugMetrics, + pub(crate) active_delete_suspect_count: u8, + pub(crate) active_delete_first_at: u64, } impl ResolverState { @@ -118,6 +120,8 @@ impl ResolverManager { resolver.last_pacing_snapshot = None; resolver.probe_attempts = 0; resolver.next_probe_at = 0; + resolver.active_delete_suspect_count = 0; + resolver.active_delete_first_at = 0; } } @@ -291,6 +295,8 @@ pub(crate) fn resolve_resolvers( }, last_pacing_snapshot: None, debug: DebugMetrics::new(debug_poll), + active_delete_suspect_count: 0, + active_delete_first_at: 0, }); } Ok(resolved) @@ -310,6 +316,8 @@ pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { resolver.last_pacing_snapshot = None; resolver.probe_attempts = 0; resolver.next_probe_at = 0; + resolver.active_delete_suspect_count = 0; + resolver.active_delete_first_at = 0; } pub(crate) fn sockaddr_storage_to_socket_addr( diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index 7128188b..42680c7d 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -13,8 +13,11 @@ use slipstream_ffi::picoquic::{ }; use slipstream_ffi::ResolverMode; use std::net::SocketAddr; +use tracing::warn; const AUTHORITATIVE_LOOP_MULTIPLIER: usize = 4; +const ACTIVE_PATH_DELETE_CONFIRM_EVENTS: u8 = 2; +const ACTIVE_PATH_DELETE_CONFIRM_WINDOW_US: u64 = 3_000_000; pub(crate) fn apply_path_mode( cnx: *mut picoquic_cnx_t, @@ -114,7 +117,13 @@ pub(crate) fn drain_path_events( PathEvent::Deleted(unique_path_id) => { if let Some(resolver) = find_resolver_by_unique_id_mut(resolvers, unique_path_id) { if resolver.is_active() { - active_path_deleted = true; + if should_confirm_active_path_delete(resolver) { + active_path_deleted = true; + reset_resolver_path(resolver); + } else { + resolver.unique_path_id = None; + } + continue; } reset_resolver_path(resolver); } @@ -125,6 +134,63 @@ pub(crate) fn drain_path_events( active_path_deleted } +fn should_confirm_active_path_delete(resolver: &mut ResolverState) -> bool { + let now = unsafe { slipstream_ffi::picoquic::picoquic_current_time() }; + let (first_at, suspect_count, confirmed) = update_active_delete_suspect( + now, + resolver.active_delete_first_at, + resolver.active_delete_suspect_count, + ); + resolver.active_delete_first_at = first_at; + resolver.active_delete_suspect_count = suspect_count; + if !confirmed { + warn!( + "active path delete observed for resolver {}; waiting for confirmation", + resolver.addr + ); + return false; + } + + true +} + +fn update_active_delete_suspect(now: u64, first_at: u64, count: u8) -> (u64, u8, bool) { + if first_at == 0 || now.saturating_sub(first_at) > ACTIVE_PATH_DELETE_CONFIRM_WINDOW_US { + return (now, 1, false); + } + + let count = count.saturating_add(1); + (first_at, count, count >= ACTIVE_PATH_DELETE_CONFIRM_EVENTS) +} + +#[cfg(test)] +mod tests { + use super::update_active_delete_suspect; + + #[test] + fn first_delete_starts_suspect_window() { + let (first_at, count, confirmed) = update_active_delete_suspect(1_000, 0, 0); + assert_eq!(first_at, 1_000); + assert_eq!(count, 1); + assert!(!confirmed); + } + + #[test] + fn second_delete_inside_window_confirms_unavailable() { + let (_, count, confirmed) = update_active_delete_suspect(2_000, 1_000, 1); + assert_eq!(count, 2); + assert!(confirmed); + } + + #[test] + fn delete_outside_window_resets_suspect_counter() { + let (first_at, count, confirmed) = update_active_delete_suspect(5_000_000, 1_000, 1); + assert_eq!(first_at, 5_000_000); + assert_eq!(count, 1); + assert!(!confirmed); + } +} + fn path_peer_addr(cnx: *mut picoquic_cnx_t, unique_path_id: u64) -> Option { let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; let ret = unsafe { picoquic_get_path_addr(cnx, unique_path_id, 2, &mut storage) }; From 57a2494f70c3fdd6fb059f92ee2719590508b9b4 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 10:52:08 +0100 Subject: [PATCH 13/21] fix(client): move path tests below runtime helpers --- crates/slipstream-client/src/runtime/path.rs | 56 ++++++++++---------- 1 file changed, 28 insertions(+), 28 deletions(-) diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index 42680c7d..1f2d096a 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -163,34 +163,6 @@ fn update_active_delete_suspect(now: u64, first_at: u64, count: u8) -> (u64, u8, (first_at, count, count >= ACTIVE_PATH_DELETE_CONFIRM_EVENTS) } -#[cfg(test)] -mod tests { - use super::update_active_delete_suspect; - - #[test] - fn first_delete_starts_suspect_window() { - let (first_at, count, confirmed) = update_active_delete_suspect(1_000, 0, 0); - assert_eq!(first_at, 1_000); - assert_eq!(count, 1); - assert!(!confirmed); - } - - #[test] - fn second_delete_inside_window_confirms_unavailable() { - let (_, count, confirmed) = update_active_delete_suspect(2_000, 1_000, 1); - assert_eq!(count, 2); - assert!(confirmed); - } - - #[test] - fn delete_outside_window_resets_suspect_counter() { - let (first_at, count, confirmed) = update_active_delete_suspect(5_000_000, 1_000, 1); - assert_eq!(first_at, 5_000_000); - assert_eq!(count, 1); - assert!(!confirmed); - } -} - fn path_peer_addr(cnx: *mut picoquic_cnx_t, unique_path_id: u64) -> Option { let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; let ret = unsafe { picoquic_get_path_addr(cnx, unique_path_id, 2, &mut storage) }; @@ -233,3 +205,31 @@ fn find_resolver_by_unique_id_mut( .iter_mut() .find(|resolver| resolver.unique_path_id == Some(unique_path_id)) } + +#[cfg(test)] +mod tests { + use super::update_active_delete_suspect; + + #[test] + fn first_delete_starts_suspect_window() { + let (first_at, count, confirmed) = update_active_delete_suspect(1_000, 0, 0); + assert_eq!(first_at, 1_000); + assert_eq!(count, 1); + assert!(!confirmed); + } + + #[test] + fn second_delete_inside_window_confirms_unavailable() { + let (_, count, confirmed) = update_active_delete_suspect(2_000, 1_000, 1); + assert_eq!(count, 2); + assert!(confirmed); + } + + #[test] + fn delete_outside_window_resets_suspect_counter() { + let (first_at, count, confirmed) = update_active_delete_suspect(5_000_000, 1_000, 1); + assert_eq!(first_at, 5_000_000); + assert_eq!(count, 1); + assert!(!confirmed); + } +} From e3b05f9d95149a883e993061defb7649b714b326 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 11:06:31 +0100 Subject: [PATCH 14/21] fix(client): gate failover on multi-signal resolver health --- crates/slipstream-client/src/dns.rs | 3 +- crates/slipstream-client/src/dns/path.rs | 43 ++++-- crates/slipstream-client/src/dns/resolver.rs | 136 ++++++++++++++++++- crates/slipstream-client/src/dns/response.rs | 4 +- crates/slipstream-client/src/runtime/path.rs | 70 ++-------- 5 files changed, 176 insertions(+), 80 deletions(-) diff --git a/crates/slipstream-client/src/dns.rs b/crates/slipstream-client/src/dns.rs index 4587cf52..71ad6b3f 100644 --- a/crates/slipstream-client/src/dns.rs +++ b/crates/slipstream-client/src/dns.rs @@ -11,6 +11,7 @@ pub(crate) use debug::{ pub(crate) use path::{add_paths, refresh_resolver_path, resolver_mode_to_c}; pub(crate) use poll::{expire_inflight_polls, send_poll_queries}; pub(crate) use resolver::{ - reset_resolver_path, sockaddr_storage_to_socket_addr, ResolverManager, ResolverState, + note_active_path_delete_signal, reset_resolver_path, should_failover_active_path, + sockaddr_storage_to_socket_addr, ResolverManager, ResolverState, }; pub(crate) use response::{handle_dns_response, DnsResponseContext}; diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index d2425277..1aef377c 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -7,10 +7,14 @@ use slipstream_ffi::picoquic::{ use slipstream_ffi::ResolverMode; use tracing::{info, warn}; -use super::resolver::{reset_resolver_path, ResolverState}; +use super::resolver::{ + clear_active_path_suspect, note_active_refresh_failure, reset_resolver_path, + should_failover_active_path, ResolverState, +}; const PATH_PROBE_INITIAL_DELAY_US: u64 = 250_000; const PATH_PROBE_MAX_DELAY_US: u64 = 10_000_000; +const PROBE_FAILURE_LOG_INTERVAL_US: u64 = 10_000_000; pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, @@ -23,8 +27,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } - resolver.active_delete_suspect_count = 0; - resolver.active_delete_first_at = 0; + resolver.last_path_unavailable_log_at = 0; + clear_active_path_suspect(resolver); return true; } resolver.unique_path_id = None; @@ -32,7 +36,13 @@ pub(crate) fn refresh_resolver_path( let peer = &resolver.storage as *const _ as *const libc::sockaddr; let path_id = unsafe { slipstream_find_path_id_by_addr(cnx, peer) }; if path_id < 0 { - if resolver.added || resolver.path_id >= 0 { + if resolver.is_active() { + let now = unsafe { picoquic_current_time() }; + note_active_refresh_failure(resolver, now); + if should_failover_active_path(resolver, now) { + reset_resolver_path(resolver); + } + } else if resolver.added || resolver.path_id >= 0 { reset_resolver_path(resolver); } return false; @@ -42,8 +52,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } - resolver.active_delete_suspect_count = 0; - resolver.active_delete_first_at = 0; + resolver.last_path_unavailable_log_at = 0; + clear_active_path_suspect(resolver); true } @@ -90,6 +100,7 @@ pub(crate) fn add_paths( if ret == 0 && path_id >= 0 { resolver.added = true; resolver.path_id = path_id; + resolver.last_probe_failure_log_at = 0; resolver.debug.path_probe_successes = resolver.debug.path_probe_successes.saturating_add(1); info!("Added path {}", resolver.addr); @@ -99,12 +110,20 @@ pub(crate) fn add_paths( resolver.probe_attempts = resolver.probe_attempts.saturating_add(1); let delay = path_probe_backoff(resolver.probe_attempts); resolver.next_probe_at = now.saturating_add(delay); - warn!( - "Failed adding path {} (attempt {}), retrying in {}ms", - resolver.addr, - resolver.probe_attempts, - delay / 1000 - ); + let should_log = resolver.probe_attempts == 1 + || resolver.probe_attempts.is_power_of_two() + || resolver.last_probe_failure_log_at == 0 + || now.saturating_sub(resolver.last_probe_failure_log_at) + >= PROBE_FAILURE_LOG_INTERVAL_US; + if should_log { + resolver.last_probe_failure_log_at = now; + warn!( + "Failed adding path {} (attempt {}), retrying in {}ms", + resolver.addr, + resolver.probe_attempts, + delay / 1000 + ); + } } if default_mode != primary_mode { diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 8e65d8a5..ee98ff8d 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -2,6 +2,7 @@ use crate::error::ClientError; use crate::pacing::{PacingBudgetSnapshot, PacingPollBudget}; use slipstream_core::state_machine::ResolverRole; use slipstream_core::{normalize_dual_stack_addr, resolve_host_port}; +use slipstream_ffi::picoquic::picoquic_current_time; use slipstream_ffi::{socket_addr_to_storage, ResolverMode, ResolverSpec}; use std::collections::HashMap; use std::net::SocketAddr; @@ -17,6 +18,18 @@ const SCORE_UNAVAILABLE_PENALTY: u64 = 2_000_000; const SCORE_TIMEOUT_PENALTY: u64 = 50_000; const SCORE_PROBE_FAILURE_PENALTY: u64 = 20_000; const SCORE_STICKINESS_BIAS: u64 = 10_000; +const PATH_UNAVAILABLE_LOG_INTERVAL_US: u64 = 5_000_000; +const ACTIVE_FAILOVER_MIN_DELETE_EVENTS: u8 = 2; +const ACTIVE_FAILOVER_MIN_REFRESH_FAILURES: u8 = 2; +const ACTIVE_FAILOVER_STALE_PROGRESS_US: u64 = 2_000_000; +const ACTIVE_FAILOVER_SUSPECT_WINDOW_US: u64 = 5_000_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ResolverHealthState { + Healthy, + Suspect, + Unavailable, +} pub(crate) struct ResolverManager { resolvers: Vec, @@ -37,13 +50,18 @@ pub(crate) struct ResolverState { pub(crate) unique_path_id: Option, pub(crate) probe_attempts: u32, pub(crate) next_probe_at: u64, + pub(crate) last_probe_failure_log_at: u64, + pub(crate) last_path_unavailable_log_at: u64, pub(crate) pending_polls: usize, pub(crate) inflight_poll_ids: HashMap, pub(crate) pacing_budget: Option, pub(crate) last_pacing_snapshot: Option, pub(crate) debug: DebugMetrics, + pub(crate) health: ResolverHealthState, pub(crate) active_delete_suspect_count: u8, + pub(crate) active_refresh_suspect_count: u8, pub(crate) active_delete_first_at: u64, + pub(crate) last_progress_at: u64, } impl ResolverState { @@ -120,8 +138,11 @@ impl ResolverManager { resolver.last_pacing_snapshot = None; resolver.probe_attempts = 0; resolver.next_probe_at = 0; + resolver.health = ResolverHealthState::Healthy; resolver.active_delete_suspect_count = 0; + resolver.active_refresh_suspect_count = 0; resolver.active_delete_first_at = 0; + resolver.last_progress_at = 0; } } @@ -287,6 +308,8 @@ pub(crate) fn resolve_resolvers( unique_path_id: None, probe_attempts: 0, next_probe_at: 0, + last_probe_failure_log_at: 0, + last_path_unavailable_log_at: 0, pending_polls: 0, inflight_poll_ids: HashMap::new(), pacing_budget: match resolver.mode { @@ -295,18 +318,28 @@ pub(crate) fn resolve_resolvers( }, last_pacing_snapshot: None, debug: DebugMetrics::new(debug_poll), + health: ResolverHealthState::Healthy, active_delete_suspect_count: 0, + active_refresh_suspect_count: 0, active_delete_first_at: 0, + last_progress_at: 0, }); } Ok(resolved) } pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { - warn!( - "Path for resolver {} became unavailable; resetting state", - resolver.addr - ); + let now = unsafe { picoquic_current_time() }; + if resolver.last_path_unavailable_log_at == 0 + || now.saturating_sub(resolver.last_path_unavailable_log_at) + >= PATH_UNAVAILABLE_LOG_INTERVAL_US + { + resolver.last_path_unavailable_log_at = now; + warn!( + "Path for resolver {} became unavailable; resetting state", + resolver.addr + ); + } resolver.added = false; resolver.path_id = -1; resolver.unique_path_id = None; @@ -316,10 +349,75 @@ pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { resolver.last_pacing_snapshot = None; resolver.probe_attempts = 0; resolver.next_probe_at = 0; + resolver.health = ResolverHealthState::Unavailable; + resolver.active_delete_suspect_count = 0; + resolver.active_refresh_suspect_count = 0; + resolver.active_delete_first_at = 0; +} + +pub(crate) fn clear_active_path_suspect(resolver: &mut ResolverState) { + resolver.health = ResolverHealthState::Healthy; resolver.active_delete_suspect_count = 0; + resolver.active_refresh_suspect_count = 0; resolver.active_delete_first_at = 0; } +pub(crate) fn note_active_path_delete_signal(resolver: &mut ResolverState, now: u64) { + if resolver.active_delete_first_at == 0 + || now.saturating_sub(resolver.active_delete_first_at) > ACTIVE_FAILOVER_SUSPECT_WINDOW_US + { + resolver.active_delete_first_at = now; + resolver.active_delete_suspect_count = 1; + resolver.active_refresh_suspect_count = 0; + resolver.health = ResolverHealthState::Suspect; + return; + } + + resolver.active_delete_suspect_count = resolver.active_delete_suspect_count.saturating_add(1); + resolver.health = ResolverHealthState::Suspect; +} + +pub(crate) fn note_active_refresh_failure(resolver: &mut ResolverState, now: u64) { + if resolver.active_delete_first_at == 0 + || now.saturating_sub(resolver.active_delete_first_at) > ACTIVE_FAILOVER_SUSPECT_WINDOW_US + { + resolver.active_delete_first_at = now; + resolver.active_delete_suspect_count = 0; + resolver.active_refresh_suspect_count = 1; + resolver.health = ResolverHealthState::Suspect; + return; + } + + resolver.active_refresh_suspect_count = resolver.active_refresh_suspect_count.saturating_add(1); + resolver.health = ResolverHealthState::Suspect; +} + +pub(crate) fn note_resolver_progress(resolver: &mut ResolverState, now: u64) { + resolver.last_progress_at = now; + if resolver.health != ResolverHealthState::Unavailable { + clear_active_path_suspect(resolver); + } +} + +pub(crate) fn should_failover_active_path(resolver: &ResolverState, now: u64) -> bool { + if resolver.health == ResolverHealthState::Unavailable { + return true; + } + if resolver.health != ResolverHealthState::Suspect { + return false; + } + if resolver.active_delete_first_at > 0 + && now.saturating_sub(resolver.active_delete_first_at) > ACTIVE_FAILOVER_SUSPECT_WINDOW_US + { + return false; + } + let stale_progress = resolver.last_progress_at == 0 + || now.saturating_sub(resolver.last_progress_at) >= ACTIVE_FAILOVER_STALE_PROGRESS_US; + stale_progress + && resolver.active_delete_suspect_count >= ACTIVE_FAILOVER_MIN_DELETE_EVENTS + && resolver.active_refresh_suspect_count >= ACTIVE_FAILOVER_MIN_REFRESH_FAILURES +} + pub(crate) fn sockaddr_storage_to_socket_addr( storage: &libc::sockaddr_storage, ) -> Result { @@ -328,7 +426,10 @@ pub(crate) fn sockaddr_storage_to_socket_addr( #[cfg(test)] mod tests { - use super::{resolve_resolvers, ResolverManager}; + use super::{ + note_active_path_delete_signal, note_active_refresh_failure, note_resolver_progress, + resolve_resolvers, should_failover_active_path, ResolverManager, + }; use slipstream_core::state_machine::ResolverRole; use slipstream_core::{AddressFamily, HostPort}; use slipstream_ffi::{ResolverMode, ResolverSpec}; @@ -461,4 +562,29 @@ mod tests { assert_eq!(manager.as_slice()[0].path_id, -1); assert_eq!(manager.as_slice()[0].unique_path_id, None); } + + #[test] + fn active_failover_requires_multi_signal_and_stale_progress() { + let resolvers = vec![ResolverSpec { + resolver: HostPort { + host: "127.0.0.1".to_string(), + port: 8853, + family: AddressFamily::V4, + }, + mode: ResolverMode::Recursive, + }]; + let mut manager = ResolverManager::from_specs(&resolvers, 900, false) + .expect("resolver manager should initialize"); + let resolver = manager.active_mut(); + + note_resolver_progress(resolver, 1_000_000); + note_active_path_delete_signal(resolver, 1_100_000); + note_active_refresh_failure(resolver, 1_200_000); + assert!(!should_failover_active_path(resolver, 1_300_000)); + + note_active_path_delete_signal(resolver, 1_400_000); + note_active_refresh_failure(resolver, 1_500_000); + assert!(!should_failover_active_path(resolver, 2_900_000)); + assert!(should_failover_active_path(resolver, 3_200_000)); + } } diff --git a/crates/slipstream-client/src/dns/response.rs b/crates/slipstream-client/src/dns/response.rs index a0f7152a..b0ed01af 100644 --- a/crates/slipstream-client/src/dns/response.rs +++ b/crates/slipstream-client/src/dns/response.rs @@ -7,7 +7,7 @@ use slipstream_ffi::picoquic::{ use slipstream_ffi::{socket_addr_to_storage, ResolverMode}; use std::net::SocketAddr; -use super::resolver::ResolverState; +use super::resolver::{note_resolver_progress, ResolverState}; use slipstream_core::normalize_dual_stack_addr; const MAX_POLL_BURST: usize = PICOQUIC_PACKET_LOOP_RECV_MAX; @@ -80,6 +80,7 @@ pub(crate) fn handle_dns_response( resolver.pending_polls = resolver.pending_polls.saturating_add(1).min(MAX_POLL_BURST); } + note_resolver_progress(resolver, current_time); } } else if let Some(response_id) = response_id { if let Some(resolver) = find_resolver_by_addr(ctx.resolvers, peer) { @@ -87,6 +88,7 @@ pub(crate) fn handle_dns_response( if resolver.mode == ResolverMode::Authoritative { resolver.inflight_poll_ids.remove(&response_id); } + note_resolver_progress(resolver, unsafe { picoquic_current_time() }); } } Ok(()) diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index 1f2d096a..e48ac6f8 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -1,5 +1,6 @@ use crate::dns::{ - record_resolver_switch, refresh_resolver_path, reset_resolver_path, resolver_mode_to_c, + note_active_path_delete_signal, record_resolver_switch, refresh_resolver_path, + reset_resolver_path, resolver_mode_to_c, should_failover_active_path, sockaddr_storage_to_socket_addr, ResolverManager, ResolverState, }; use crate::error::ClientError; @@ -16,8 +17,6 @@ use std::net::SocketAddr; use tracing::warn; const AUTHORITATIVE_LOOP_MULTIPLIER: usize = 4; -const ACTIVE_PATH_DELETE_CONFIRM_EVENTS: u8 = 2; -const ACTIVE_PATH_DELETE_CONFIRM_WINDOW_US: u64 = 3_000_000; pub(crate) fn apply_path_mode( cnx: *mut picoquic_cnx_t, @@ -117,10 +116,16 @@ pub(crate) fn drain_path_events( PathEvent::Deleted(unique_path_id) => { if let Some(resolver) = find_resolver_by_unique_id_mut(resolvers, unique_path_id) { if resolver.is_active() { - if should_confirm_active_path_delete(resolver) { + let now = unsafe { slipstream_ffi::picoquic::picoquic_current_time() }; + note_active_path_delete_signal(resolver, now); + if should_failover_active_path(resolver, now) { active_path_deleted = true; reset_resolver_path(resolver); } else { + warn!( + "active path delete observed for resolver {}; waiting for progress-confirmed failover", + resolver.addr + ); resolver.unique_path_id = None; } continue; @@ -134,35 +139,6 @@ pub(crate) fn drain_path_events( active_path_deleted } -fn should_confirm_active_path_delete(resolver: &mut ResolverState) -> bool { - let now = unsafe { slipstream_ffi::picoquic::picoquic_current_time() }; - let (first_at, suspect_count, confirmed) = update_active_delete_suspect( - now, - resolver.active_delete_first_at, - resolver.active_delete_suspect_count, - ); - resolver.active_delete_first_at = first_at; - resolver.active_delete_suspect_count = suspect_count; - if !confirmed { - warn!( - "active path delete observed for resolver {}; waiting for confirmation", - resolver.addr - ); - return false; - } - - true -} - -fn update_active_delete_suspect(now: u64, first_at: u64, count: u8) -> (u64, u8, bool) { - if first_at == 0 || now.saturating_sub(first_at) > ACTIVE_PATH_DELETE_CONFIRM_WINDOW_US { - return (now, 1, false); - } - - let count = count.saturating_add(1); - (first_at, count, count >= ACTIVE_PATH_DELETE_CONFIRM_EVENTS) -} - fn path_peer_addr(cnx: *mut picoquic_cnx_t, unique_path_id: u64) -> Option { let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; let ret = unsafe { picoquic_get_path_addr(cnx, unique_path_id, 2, &mut storage) }; @@ -205,31 +181,3 @@ fn find_resolver_by_unique_id_mut( .iter_mut() .find(|resolver| resolver.unique_path_id == Some(unique_path_id)) } - -#[cfg(test)] -mod tests { - use super::update_active_delete_suspect; - - #[test] - fn first_delete_starts_suspect_window() { - let (first_at, count, confirmed) = update_active_delete_suspect(1_000, 0, 0); - assert_eq!(first_at, 1_000); - assert_eq!(count, 1); - assert!(!confirmed); - } - - #[test] - fn second_delete_inside_window_confirms_unavailable() { - let (_, count, confirmed) = update_active_delete_suspect(2_000, 1_000, 1); - assert_eq!(count, 2); - assert!(confirmed); - } - - #[test] - fn delete_outside_window_resets_suspect_counter() { - let (first_at, count, confirmed) = update_active_delete_suspect(5_000_000, 1_000, 1); - assert_eq!(first_at, 5_000_000); - assert_eq!(count, 1); - assert!(!confirmed); - } -} From 8c3424db18dba91b93b5559dcff077c0b1c7a643 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 12:39:02 +0100 Subject: [PATCH 15/21] fix(client): keep recursive standby paths inactive --- crates/slipstream-client/src/dns/path.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 1aef377c..25229962 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -15,6 +15,7 @@ use super::resolver::{ const PATH_PROBE_INITIAL_DELAY_US: u64 = 250_000; const PATH_PROBE_MAX_DELAY_US: u64 = 10_000_000; const PROBE_FAILURE_LOG_INTERVAL_US: u64 = 10_000_000; +const SKIP_STANDBY_RECURSIVE_PATH_PROBES: bool = true; pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, @@ -75,6 +76,9 @@ pub(crate) fn add_paths( let mut default_mode = primary_mode; for resolver in resolvers.iter_mut().skip(1) { + if SKIP_STANDBY_RECURSIVE_PATH_PROBES && resolver.mode == ResolverMode::Recursive { + continue; + } if resolver.added { continue; } From 99acd5ffaf3bd4206cc2637b45cbfe6f39f6cd55 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Mon, 16 Feb 2026 14:07:05 +0100 Subject: [PATCH 16/21] fix(client): preserve mixed-mode polling on standby authoritative path --- crates/slipstream-client/src/runtime.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 80bb1fd0..05e3d158 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -595,7 +595,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if !refresh_resolver_path(cnx, resolver) { continue; } - if !resolver.is_active() { + if !resolver.is_active() && resolver.mode == ResolverMode::Recursive { continue; } let pending_for_sleep = match resolver.mode { @@ -928,7 +928,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } watchdog.set_phase(PHASE_POLL_QUERIES); for resolver in resolver_manager.as_mut_slice().iter_mut() { - if !resolver.is_active() { + if !resolver.is_active() && resolver.mode == ResolverMode::Recursive { continue; } if !refresh_resolver_path(cnx, resolver) { From 26882fcb2633759f2a7700208e4426e1092d0e2d Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 19:19:07 +0100 Subject: [PATCH 17/21] refactor(client): reconcile runtime helpers for minimal branch --- crates/slipstream-client/src/runtime.rs | 73 ++++---------------- crates/slipstream-client/src/runtime/path.rs | 12 ---- 2 files changed, 13 insertions(+), 72 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 05e3d158..46dbc612 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -1,35 +1,16 @@ mod path; mod setup; -<<<<<<< HEAD use self::path::{ apply_path_mode, drain_path_events, fetch_path_quality, find_resolver_by_addr_mut, - loop_burst_total, path_poll_burst_max, -======= -use self::actions::{poll_authoritative_resolver, poll_recursive_resolver, PollDispatch}; -use self::deadlock::AcceptorSaturationTracker; -use self::decision::{reconnect_due_to_acceptor_deadlock, reconnect_due_to_active_path_loss}; -use self::health::{ - compute_last_enqueue_ms, should_log_flow_blocked, should_log_health, - should_reconnect_for_handshake_stall, should_reconnect_for_resolver_stall, -}; -use self::loop_policy::compute_loop_sleep_policy; -use self::path::{ - apply_path_mode, drain_path_events, fetch_and_record_path_quality, find_resolver_by_addr_mut, - loop_burst_total, maybe_switch_active_resolver, ->>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) + loop_burst_total, maybe_switch_active_resolver, path_poll_burst_max, }; use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ add_paths, expire_inflight_polls, handle_dns_response, maybe_report_debug, record_resolver_switch, refresh_resolver_path, resolver_mode_to_c, -<<<<<<< HEAD resolver_switch_reason_catalog, send_poll_queries, sockaddr_storage_to_socket_addr, DnsResponseContext, ResolverManager, ResolverSwitchReason, -======= - resolver_switch_reason_catalog, sockaddr_storage_to_socket_addr, DnsResponseContext, - ResolverManager, ResolverSwitchReason, ->>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -234,40 +215,20 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver dropped } -<<<<<<< HEAD -fn maybe_switch_active_resolver( - resolver_manager: &mut ResolverManager, +fn should_reconnect_for_handshake_stall( + ready: bool, current_time: u64, - preferred_startup_resolver_index: &mut usize, -) { - if let Some((from_index, to_index, reason)) = resolver_manager.maybe_select_active(current_time) - { - *preferred_startup_resolver_index = to_index; - record_resolver_switch( - resolver_manager.as_mut_slice(), - Some(from_index), - to_index, - reason, - ); - unsafe { - let mode = resolver_manager.active().mode; - slipstream_set_default_path_mode(resolver_mode_to_c(mode)); - } -======= -fn drain_commands_by_connection_state( - cnx: *mut picoquic_cnx_t, - state_ptr: *mut ClientState, - command_rx: &mut mpsc::UnboundedReceiver, -) { - // Only process application commands after the QUIC handshake - // completes. During reconnect the acceptor may queue NewStream - // commands while the connection is still in initial state; - // processing them before ready triggers picoquic errors (0xc). - if unsafe { (*state_ptr).is_ready() } { - drain_commands(cnx, state_ptr, command_rx); + connect_started_at: u64, + timeout_us: u64, +) -> Option { + if ready || connect_started_at == 0 { + return None; + } + let elapsed = current_time.saturating_sub(connect_started_at); + if elapsed >= timeout_us { + Some(elapsed) } else { - let _ = drain_disconnected_commands(command_rx); ->>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) + None } } @@ -717,7 +678,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { &mut resolver_manager, current_time, &mut preferred_startup_resolver_index, -<<<<<<< HEAD ); let streams_len = unsafe { (*state_ptr).streams_len() }; let active_path_ready = resolver_manager.active().added; @@ -728,13 +688,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { ); break; } -======= - state_ptr, - ACTIVE_PATH_LOSS_RECONNECT_STREAMS, - ) - { - break; ->>>>>>> e3523a8 (fix(client): rotate startup resolver after handshake stall) } let reaped_half_closed = unsafe { let now = picoquic_current_time(); diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index e48ac6f8..854d0961 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -50,18 +50,6 @@ pub(crate) fn fetch_path_quality( quality } -pub(crate) fn fetch_and_record_path_quality( - cnx: *mut picoquic_cnx_t, - resolver: &mut ResolverState, -) -> slipstream_ffi::picoquic::picoquic_path_quality_t { - let quality = fetch_path_quality(cnx, resolver); - resolver.debug.path_rtt_us = quality.rtt; - resolver.debug.path_cwnd = quality.cwin; - resolver.debug.path_bytes_in_transit = quality.bytes_in_transit; - resolver.debug.path_pacing_rate = quality.pacing_rate; - quality -} - pub(crate) fn maybe_switch_active_resolver( resolver_manager: &mut ResolverManager, current_time: u64, From e4579ae7eeb96137c5be1f4d04c46d600f97155c Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Tue, 17 Feb 2026 22:08:34 +0100 Subject: [PATCH 18/21] fix(client): harden watchdog against transient select stalls --- crates/slipstream-client/src/runtime.rs | 30 ++++++++++++++++++++++--- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 46dbc612..118e7227 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -76,6 +76,8 @@ const ACCEPTOR_SATURATED_TIMEOUT_US: u64 = 30_000_000; const HEALTH_LOG_INTERVAL_US: u64 = 300_000_000; const WATCHDOG_STALE_SECS: u64 = 15; const WATCHDOG_CHECK_INTERVAL: Duration = Duration::from_secs(3); +const WATCHDOG_ABORT_STRIKES: u32 = 3; +const WATCHDOG_SELECT_ABORT_SECS: u64 = 45; const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32; /// Watchdog that runs on a separate OS thread (not tokio) to detect when the @@ -128,6 +130,7 @@ impl Watchdog { .name("watchdog".into()) .spawn(move || { let mut last_check = Instant::now(); + let mut stale_strikes = 0u32; while al.load(Ordering::Relaxed) { std::thread::sleep(WATCHDOG_CHECK_INTERVAL); if !al.load(Ordering::Relaxed) { @@ -136,6 +139,7 @@ impl Watchdog { let now_instant = Instant::now(); let ts = hb.load(Ordering::Relaxed); if ts == 0 { + stale_strikes = 0; last_check = now_instant; continue; } @@ -149,6 +153,7 @@ impl Watchdog { if stale_us > WATCHDOG_STALE_SECS * 1_000_000 && own_sleep_us > expected_sleep_us * 3 { + stale_strikes = 0; let stuck_phase = ph.load(Ordering::Relaxed); eprintln!( "WATCHDOG: VPS suspend detected ({:.1}s gap, own sleep {:.1}s), \ @@ -162,14 +167,33 @@ impl Watchdog { continue; } if stale_us > WATCHDOG_STALE_SECS * 1_000_000 { + stale_strikes = stale_strikes.saturating_add(1); let stuck_phase = ph.load(Ordering::Relaxed); + let phase_name = phase_name(stuck_phase); + let allow_abort = if stuck_phase == PHASE_SELECT { + stale_us >= WATCHDOG_SELECT_ABORT_SECS * 1_000_000 + } else { + true + }; + if allow_abort && stale_strikes >= WATCHDOG_ABORT_STRIKES { + eprintln!( + "WATCHDOG: main loop stalled for {:.1}s at phase {} ({}), strikes={}, aborting process", + stale_us as f64 / 1_000_000.0, + stuck_phase, + phase_name, + stale_strikes, + ); + std::process::abort(); + } eprintln!( - "WATCHDOG: main loop stalled for {:.1}s at phase {} ({}), aborting process", + "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting", stale_us as f64 / 1_000_000.0, stuck_phase, - phase_name(stuck_phase), + phase_name, + stale_strikes, ); - std::process::abort(); + } else { + stale_strikes = 0; } } }) From 722795722342578f05e3becb6f1f2f1afd30d0ed Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Wed, 18 Feb 2026 08:59:48 +0100 Subject: [PATCH 19/21] fix(client): avoid aborting watchdog in select phase --- crates/slipstream-client/src/runtime.rs | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 118e7227..73bacb98 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -77,13 +77,15 @@ const HEALTH_LOG_INTERVAL_US: u64 = 300_000_000; const WATCHDOG_STALE_SECS: u64 = 15; const WATCHDOG_CHECK_INTERVAL: Duration = Duration::from_secs(3); const WATCHDOG_ABORT_STRIKES: u32 = 3; -const WATCHDOG_SELECT_ABORT_SECS: u64 = 45; const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32; /// Watchdog that runs on a separate OS thread (not tokio) to detect when the /// single-threaded tokio runtime freezes (e.g. a picoquic C FFI call hangs). /// If the main loop hasn't updated the heartbeat for WATCHDOG_STALE_SECS, -/// the watchdog aborts the process so systemd can restart it. +/// the watchdog emits warnings and only aborts in non-select phases. +/// +/// `PHASE_SELECT` may legitimately appear stale under host scheduler jitter, +/// so aborting there creates false outages. struct Watchdog { heartbeat: Arc, phase: Arc, @@ -170,12 +172,17 @@ impl Watchdog { stale_strikes = stale_strikes.saturating_add(1); let stuck_phase = ph.load(Ordering::Relaxed); let phase_name = phase_name(stuck_phase); - let allow_abort = if stuck_phase == PHASE_SELECT { - stale_us >= WATCHDOG_SELECT_ABORT_SECS * 1_000_000 - } else { - true - }; - if allow_abort && stale_strikes >= WATCHDOG_ABORT_STRIKES { + if stuck_phase == PHASE_SELECT { + eprintln!( + "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting (select phase never aborts)", + stale_us as f64 / 1_000_000.0, + stuck_phase, + phase_name, + stale_strikes, + ); + continue; + } + if stale_strikes >= WATCHDOG_ABORT_STRIKES { eprintln!( "WATCHDOG: main loop stalled for {:.1}s at phase {} ({}), strikes={}, aborting process", stale_us as f64 / 1_000_000.0, From 6ed9143d1ccbb514e4b3f60f821c6a7415dc9e5a Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Wed, 18 Feb 2026 14:32:54 +0100 Subject: [PATCH 20/21] fix(client): bound select-phase watchdog stalls --- crates/slipstream-client/src/runtime.rs | 45 +++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 73bacb98..d51e99f8 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -77,6 +77,8 @@ const HEALTH_LOG_INTERVAL_US: u64 = 300_000_000; const WATCHDOG_STALE_SECS: u64 = 15; const WATCHDOG_CHECK_INTERVAL: Duration = Duration::from_secs(3); const WATCHDOG_ABORT_STRIKES: u32 = 3; +const WATCHDOG_SELECT_RECONNECT_SECS: u64 = 30; +const WATCHDOG_SELECT_ABORT_SECS: u64 = 180; const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32; /// Watchdog that runs on a separate OS thread (not tokio) to detect when the @@ -85,11 +87,12 @@ const ACTIVE_PATH_LOSS_RECONNECT_STREAMS: usize = 32; /// the watchdog emits warnings and only aborts in non-select phases. /// /// `PHASE_SELECT` may legitimately appear stale under host scheduler jitter, -/// so aborting there creates false outages. +/// so we first request reconnect; abort is only a last-resort cap. struct Watchdog { heartbeat: Arc, phase: Arc, alive: Arc, + select_reconnect_requested: Arc, _handle: std::thread::JoinHandle<()>, } @@ -125,9 +128,11 @@ impl Watchdog { let heartbeat = Arc::new(AtomicU64::new(0)); let phase = Arc::new(AtomicU32::new(0)); let alive = Arc::new(AtomicBool::new(true)); + let select_reconnect_requested = Arc::new(AtomicBool::new(false)); let hb = Arc::clone(&heartbeat); let ph = Arc::clone(&phase); let al = Arc::clone(&alive); + let reconnect_flag = Arc::clone(&select_reconnect_requested); let handle = std::thread::Builder::new() .name("watchdog".into()) .spawn(move || { @@ -173,8 +178,26 @@ impl Watchdog { let stuck_phase = ph.load(Ordering::Relaxed); let phase_name = phase_name(stuck_phase); if stuck_phase == PHASE_SELECT { + if stale_us >= WATCHDOG_SELECT_RECONNECT_SECS * 1_000_000 { + let first_request = !reconnect_flag.swap(true, Ordering::Relaxed); + if first_request { + eprintln!( + "WATCHDOG: select phase stale for {:.1}s; requesting reconnect", + stale_us as f64 / 1_000_000.0, + ); + } + } + if stale_us >= WATCHDOG_SELECT_ABORT_SECS * 1_000_000 { + eprintln!( + "WATCHDOG: select phase stalled for {:.1}s (phase {} / {}), aborting as last resort", + stale_us as f64 / 1_000_000.0, + stuck_phase, + phase_name, + ); + std::process::abort(); + } eprintln!( - "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting (select phase never aborts)", + "WATCHDOG: stale heartbeat {:.1}s at phase {} ({}), strikes={}, waiting (select phase uses reconnect-first policy)", stale_us as f64 / 1_000_000.0, stuck_phase, phase_name, @@ -209,6 +232,7 @@ impl Watchdog { heartbeat, phase, alive, + select_reconnect_requested, _handle: handle, } } @@ -221,6 +245,11 @@ impl Watchdog { fn set_phase(&self, p: u32) { self.phase.store(p, Ordering::Relaxed); } + + fn take_select_reconnect_requested(&self) -> bool { + self.select_reconnect_requested + .swap(false, Ordering::Relaxed) + } } impl Drop for Watchdog { @@ -467,6 +496,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { watchdog.pet(); watchdog.set_phase(PHASE_DRAIN_COMMANDS); let current_time = unsafe { picoquic_current_time() }; + if watchdog.take_select_reconnect_requested() { + warn!( + "watchdog requested reconnect after select-phase stall; recycling connection" + ); + break; + } // Only process application commands after the QUIC handshake // completes. During reconnect the acceptor may queue NewStream // commands while the connection is still in initial state; @@ -694,6 +729,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { ); } watchdog.pet(); + if watchdog.take_select_reconnect_requested() { + warn!( + "watchdog requested reconnect after select-phase stall; recycling connection" + ); + break; + } watchdog.set_phase(PHASE_POST_DRAIN); if unsafe { (*state_ptr).is_ready() } { From 9f15dfa51708a054a3dc474350bed308c1218310 Mon Sep 17 00:00:00 2001 From: Night Owl Nerd <256460992+nightowlnerd@users.noreply.github.com> Date: Wed, 18 Feb 2026 16:54:30 +0100 Subject: [PATCH 21/21] fix(client): rotate startup resolver on pre-ready reconnects --- crates/slipstream-client/src/runtime.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index d51e99f8..b7bf1b2d 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -484,6 +484,8 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut acceptor_saturated_since: u64 = 0; let mut acceptor_saturated_max: usize = 0; let mut acceptor_saturated_bytes: u64 = 0; + let mut reached_ready = false; + let mut rotated_on_handshake_stall = false; let watchdog = Watchdog::spawn(); // Clear closing flag that picoquic_free (QuicGuard::drop) may have @@ -530,6 +532,9 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } let ready = unsafe { (*state_ptr).is_ready() }; + if ready { + reached_ready = true; + } if let Some(stall_us) = should_reconnect_for_handshake_stall( ready, current_time, @@ -548,6 +553,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { to_index, ResolverSwitchReason::HandshakeStall, ); + rotated_on_handshake_stall = true; preferred_startup_resolver_index = to_index; warn!( "handshake stall for {:.1}s on startup resolver {}; rotating startup resolver to {} and reconnecting", @@ -1138,6 +1144,19 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { picoquic_close(cnx, 0); } + if !reached_ready && !rotated_on_handshake_stall && resolver_manager.as_slice().len() > 1 { + let from_index = resolver_manager.active_index(); + let to_index = (from_index + 1) % resolver_manager.as_slice().len(); + let from_addr = resolver_manager.as_slice()[from_index].addr; + let to_addr = resolver_manager.as_slice()[to_index].addr; + preferred_startup_resolver_index = to_index; + warn!( + "connection closed before ready on startup resolver {}; rotating startup resolver to {} for next attempt", + from_addr, + to_addr, + ); + } + unsafe { (*state_ptr).reset_for_reconnect(); }