diff --git a/crates/slipstream-client/src/dns.rs b/crates/slipstream-client/src/dns.rs index 0b077b32..71ad6b3f 100644 --- a/crates/slipstream-client/src/dns.rs +++ b/crates/slipstream-client/src/dns.rs @@ -4,10 +4,14 @@ mod poll; mod resolver; mod response; -pub(crate) use debug::maybe_report_debug; +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::{ - reset_resolver_path, resolve_resolvers, sockaddr_storage_to_socket_addr, 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/debug.rs b/crates/slipstream-client/src/dns/debug.rs index 4b60a20f..03e9f2f0 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,99 @@ 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, + HandshakeStall, + LossSpike, + LatencyRegression, + PathUnavailable, + CooldownExpired, +} + +pub(crate) fn resolver_switch_reason_catalog() -> &'static [ResolverSwitchReason] { + static REASONS: [ResolverSwitchReason; 9] = [ + ResolverSwitchReason::StartupPrimary, + ResolverSwitchReason::ManualOverride, + ResolverSwitchReason::ProbeRecovery, + ResolverSwitchReason::TimeoutStreakExceeded, + ResolverSwitchReason::HandshakeStall, + ResolverSwitchReason::LossSpike, + ResolverSwitchReason::LatencyRegression, + ResolverSwitchReason::PathUnavailable, + ResolverSwitchReason::CooldownExpired, + ]; + &REASONS +} + +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::HandshakeStall => "handshake_stall", + 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 +160,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 +208,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 +222,66 @@ 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::HandshakeStall.as_str(), + "handshake_stall" + ); + 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..25229962 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -7,10 +7,15 @@ 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; +const SKIP_STANDBY_RECURSIVE_PATH_PROBES: bool = true; pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, @@ -23,6 +28,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } + resolver.last_path_unavailable_log_at = 0; + clear_active_path_suspect(resolver); return true; } resolver.unique_path_id = None; @@ -30,7 +37,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; @@ -40,6 +53,8 @@ pub(crate) fn refresh_resolver_path( if resolver.path_id != path_id { resolver.path_id = path_id; } + resolver.last_path_unavailable_log_at = 0; + clear_active_path_suspect(resolver); true } @@ -61,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; } @@ -86,18 +104,30 @@ 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); 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); - 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/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..ee98ff8d 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -1,37 +1,277 @@ 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; 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; +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, + active_index: usize, + last_switch_at: u64, + candidate_index: Option, + candidate_wins: u8, +} pub(crate) struct ResolverState { pub(crate) addr: SocketAddr, 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, 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 { 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 ) } + + pub(crate) fn is_active(&self) -> bool { + self.role == ResolverRole::Active + } +} + +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, + last_switch_at: 0, + candidate_index: None, + candidate_wins: 0, + }) + } + + pub(crate) fn active_index(&self) -> usize { + 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 = 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; + 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; + } + } + + pub(crate) fn active_mut(&mut self) -> &mut ResolverState { + &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 + } + + 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( @@ -58,11 +298,18 @@ 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 }, + 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 { @@ -71,16 +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; @@ -90,6 +349,73 @@ 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( @@ -100,7 +426,11 @@ pub(crate) fn sockaddr_storage_to_socket_addr( #[cfg(test)] mod tests { - use super::resolve_resolvers; + 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}; @@ -130,4 +460,131 @@ 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); + } + + #[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); + } + + #[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, 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); + } + + #[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.rs b/crates/slipstream-client/src/runtime.rs index 57dacfa6..b7bf1b2d 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -3,13 +3,14 @@ mod setup; use self::path::{ apply_path_mode, drain_path_events, fetch_path_quality, find_resolver_by_addr_mut, - loop_burst_total, path_poll_burst_max, + 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, - refresh_resolver_path, resolve_resolvers, resolver_mode_to_c, send_poll_queries, - sockaddr_storage_to_socket_addr, DnsResponseContext, + 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}; @@ -58,6 +59,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. @@ -71,15 +76,23 @@ 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_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 /// 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 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<()>, } @@ -115,13 +128,16 @@ 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 || { 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) { @@ -130,6 +146,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; } @@ -143,6 +160,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), \ @@ -156,14 +174,56 @@ 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); + 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 uses reconnect-first policy)", + 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, + 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; } } }) @@ -172,6 +232,7 @@ impl Watchdog { heartbeat, phase, alive, + select_reconnect_requested, _handle: handle, } } @@ -184,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 { @@ -209,7 +275,25 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver dropped } +fn should_reconnect_for_handshake_stall( + ready: bool, + current_time: u64, + 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 { + None + } +} + 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?; @@ -272,16 +356,24 @@ 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 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)?; + 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(), + None, + active_index, + ResolverSwitchReason::StartupPrimary, + ); 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, @@ -320,7 +412,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. @@ -334,12 +426,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( @@ -357,7 +451,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 _); @@ -377,8 +471,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; @@ -388,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 @@ -400,6 +498,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; @@ -428,6 +532,43 @@ 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, + 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::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", + 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; @@ -438,18 +579,42 @@ 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); + 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() }; + 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 and no standby path ready; reconnecting to limit reset storm", + streams_len, + ); + break; + } + } - for resolver in resolvers.iter_mut() { + for resolver in resolver_manager.as_mut_slice().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); + } } } @@ -459,13 +624,20 @@ 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; } + if !resolver.is_active() && resolver.mode == ResolverMode::Recursive { + continue; + } 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() @@ -524,7 +696,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 { @@ -563,6 +735,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() } { @@ -571,7 +749,24 @@ 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); + 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() }; + 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 and no standby path ready; 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) @@ -665,8 +860,11 @@ 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() { - if resolver.mode == ResolverMode::Recursive && resolver.added { + for resolver in resolver_manager.as_mut_slice().iter_mut() { + if resolver.is_active() + && resolver.mode == ResolverMode::Recursive + && resolver.added + { resolver.pending_polls = resolver.pending_polls.max(1); } } @@ -680,7 +878,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 = @@ -715,6 +915,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } + 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 }; let streams_len = unsafe { (*state_ptr).streams_len() }; @@ -752,13 +958,20 @@ 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 !resolver.is_active() && resolver.mode == ResolverMode::Recursive { + continue; + } if !refresh_resolver_path(cnx, resolver) { continue; } 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) @@ -866,7 +1079,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; @@ -879,6 +1092,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 @@ -927,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(); } diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index 966a4e75..854d0961 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -1,17 +1,20 @@ use crate::dns::{ - refresh_resolver_path, reset_resolver_path, resolver_mode_to_c, - sockaddr_storage_to_socket_addr, ResolverState, + 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; 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; +use tracing::warn; const AUTHORITATIVE_LOOP_MULTIPLIER: usize = 4; @@ -47,18 +50,40 @@ pub(crate) fn fetch_path_quality( 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], 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 +103,28 @@ 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() { + 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; + } reset_resolver_path(resolver); } } } } + + active_path_deleted } fn path_peer_addr(cnx: *mut picoquic_cnx_t, unique_path_id: u64) -> Option { 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"); + } +}