From 8fc3488c151a44159200a54a7f20cfa12d6bebe0 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 10:39:31 +0300 Subject: [PATCH 01/19] Stabilize DNS tunnel QUIC polling --- crates/slipstream-client/src/dns/path.rs | 15 ++- crates/slipstream-client/src/dns/resolver.rs | 14 +++ crates/slipstream-client/src/main.rs | 30 +++++- crates/slipstream-client/src/runtime.rs | 96 +++++++++++++++++-- .../slipstream-client/src/streams/callback.rs | 16 +++- crates/slipstream-ffi/src/lib.rs | 1 + crates/slipstream-ffi/src/picoquic.rs | 1 + crates/slipstream-server/src/main.rs | 23 +++++ crates/slipstream-server/src/server.rs | 22 ++++- .../tests/cert_pinning_e2e.rs | 1 + .../tests/epipe_reset_e2e.rs | 5 +- .../tests/flow_control_e2e.rs | 1 + crates/slipstream-server/tests/idle_gc_e2e.rs | 11 ++- .../tests/restart_reconnect_e2e.rs | 2 + .../tests/stream_limit_e2e.rs | 2 + crates/slipstream-server/tests/support/mod.rs | 4 + .../tests/target_read_error_reset_e2e.rs | 1 + .../tests/udp_fallback_e2e.rs | 1 + docs/config.md | 8 +- docs/sip003.md | 10 +- docs/usage.md | 7 +- 21 files changed, 248 insertions(+), 23 deletions(-) diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 5042aa91..ff0b7979 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -11,6 +11,8 @@ use super::resolver::{reset_resolver_path, ResolverState}; const PATH_PROBE_INITIAL_DELAY_US: u64 = 250_000; const PATH_PROBE_MAX_DELAY_US: u64 = 10_000_000; +const PATH_PROBE_DISABLE_AFTER_ATTEMPTS: u32 = 5; +const PATH_PROBE_DISABLE_US: u64 = 30_000_000; pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, @@ -64,6 +66,9 @@ pub(crate) fn add_paths( if resolver.added { continue; } + if resolver.disabled_until > now { + continue; + } if resolver.next_probe_at > now { continue; } @@ -90,12 +95,18 @@ pub(crate) fn add_paths( continue; } resolver.probe_attempts = resolver.probe_attempts.saturating_add(1); - let delay = path_probe_backoff(resolver.probe_attempts); + let attempt = resolver.probe_attempts; + let mut delay = path_probe_backoff(resolver.probe_attempts); + if resolver.probe_attempts >= PATH_PROBE_DISABLE_AFTER_ATTEMPTS { + resolver.disabled_until = now.saturating_add(PATH_PROBE_DISABLE_US); + resolver.probe_attempts = 0; + delay = PATH_PROBE_DISABLE_US; + } resolver.next_probe_at = now.saturating_add(delay); warn!( "Failed adding path {} (attempt {}), retrying in {}ms", resolver.addr, - resolver.probe_attempts, + attempt, delay / 1000 ); } diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 320421e2..9b11d68b 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -41,6 +41,11 @@ pub(crate) struct ResolverState { pub(crate) unique_path_id: Option, pub(crate) probe_attempts: u32, pub(crate) next_probe_at: u64, + pub(crate) disabled_until: u64, + pub(crate) last_health_check_at: u64, + pub(crate) last_health_send_packets: u64, + pub(crate) last_health_dns_responses: u64, + pub(crate) last_active_poll_kick_at: u64, pub(crate) pending_polls: usize, pub(crate) inflight_poll_ids: HashMap, pub(crate) pacing_budget: Option, @@ -87,6 +92,11 @@ pub(crate) fn resolve_resolvers( unique_path_id: if is_primary { Some(0) } else { None }, probe_attempts: 0, next_probe_at: 0, + disabled_until: 0, + last_health_check_at: 0, + last_health_send_packets: 0, + last_health_dns_responses: 0, + last_active_poll_kick_at: 0, pending_polls: 0, inflight_poll_ids: HashMap::new(), pacing_budget: match resolver.mode { @@ -114,6 +124,10 @@ pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { resolver.last_pacing_snapshot = None; resolver.probe_attempts = 0; resolver.next_probe_at = 0; + resolver.last_health_check_at = 0; + resolver.last_health_send_packets = 0; + resolver.last_health_dns_responses = 0; + resolver.last_active_poll_kick_at = 0; } pub(crate) fn sockaddr_storage_to_socket_addr( diff --git a/crates/slipstream-client/src/main.rs b/crates/slipstream-client/src/main.rs index fce871a9..ac8e2548 100644 --- a/crates/slipstream-client/src/main.rs +++ b/crates/slipstream-client/src/main.rs @@ -52,8 +52,15 @@ struct Args { domain: Option, #[arg(long = "cert", value_name = "PATH")] cert: Option, - #[arg(long = "keep-alive-interval", short = 't', default_value_t = 400)] + #[arg( + long = "keep-alive-interval", + short = 't', + default_value_t = 400, + help = "Send keep alive pings at this interval in milliseconds (disabled: 0)" + )] keep_alive_interval: u16, + #[arg(long = "quic-idle-timeout-seconds", default_value_t = 120)] + quic_idle_timeout_seconds: u64, #[arg(long = "debug-poll")] debug_poll: bool, #[arg(long = "debug-streams")] @@ -170,6 +177,19 @@ fn main() { ); keep_alive_override.unwrap_or(args.keep_alive_interval) }; + let quic_idle_timeout_seconds = if cli_provided(&matches, "quic_idle_timeout_seconds") { + args.quic_idle_timeout_seconds + } else { + sip003::last_option_value(&sip003_env.plugin_options, "quic-idle-timeout-seconds") + .map(|value| { + unwrap_or_exit( + parse_quic_idle_timeout_seconds(&value), + "SIP003 env error", + 2, + ) + }) + .unwrap_or(args.quic_idle_timeout_seconds) + }; let config = ClientConfig { tcp_listen_host: &tcp_listen_host, @@ -180,6 +200,7 @@ fn main() { domain: &domain, cert: cert.as_deref(), keep_alive_interval: keep_alive_interval as usize, + quic_idle_timeout_seconds, debug_poll: args.debug_poll, debug_streams: args.debug_streams, }; @@ -344,6 +365,13 @@ fn parse_keep_alive_interval(options: &[sip003::Sip003Option]) -> Result Result { + let trimmed = value.trim(); + trimmed + .parse::() + .map_err(|_| format!("Invalid quic-idle-timeout-seconds value: {}", trimmed)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index ca4dc505..1bd79ddc 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, PeerAddrMode, + refresh_resolver_path, reset_resolver_path, resolve_resolvers, resolver_mode_to_c, + send_poll_queries, sockaddr_storage_to_socket_addr, DnsResponseContext, PeerAddrMode, }; use crate::error::ClientError; use crate::pacing::{cwnd_target_polls, inflight_packet_estimate}; @@ -27,10 +27,11 @@ use slipstream_ffi::{ picoquic_create_client_cnx, picoquic_current_time, picoquic_disable_keep_alive, picoquic_enable_keep_alive, picoquic_enable_path_callbacks, picoquic_enable_path_callbacks_default, picoquic_get_next_wake_delay, - picoquic_prepare_next_packet_ex, picoquic_set_callback, slipstream_has_ready_stream, - slipstream_is_flow_blocked, slipstream_mixed_cc_algorithm, slipstream_set_cc_override, - slipstream_set_default_path_mode, PICOQUIC_CONNECTION_ID_MAX_SIZE, - PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, PICOQUIC_PACKET_LOOP_SEND_MAX, + picoquic_prepare_next_packet_ex, picoquic_set_callback, picoquic_set_default_idle_timeout, + slipstream_has_ready_stream, slipstream_is_flow_blocked, slipstream_mixed_cc_algorithm, + slipstream_set_cc_override, slipstream_set_default_path_mode, + PICOQUIC_CONNECTION_ID_MAX_SIZE, PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, + PICOQUIC_PACKET_LOOP_SEND_MAX, }, socket_addr_to_storage, take_crypto_errors, ClientConfig, QuicGuard, ResolverMode, }; @@ -49,6 +50,10 @@ const DNS_POLL_SLICE_US: u64 = 50_000; const RECONNECT_SLEEP_MIN_MS: u64 = 250; const RECONNECT_SLEEP_MAX_MS: u64 = 5_000; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; +const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; +const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; +const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; +const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { let mut dropped = 0usize; @@ -61,6 +66,72 @@ fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver dropped } +fn maybe_disable_no_progress_path( + resolver: &mut crate::dns::ResolverState, + now: u64, + active_streams: bool, +) { + if !active_streams || !resolver.added { + resolver.last_health_check_at = now; + resolver.last_health_send_packets = resolver.debug.send_packets; + resolver.last_health_dns_responses = resolver.debug.dns_responses; + return; + } + if resolver.last_health_check_at == 0 { + resolver.last_health_check_at = now; + resolver.last_health_send_packets = resolver.debug.send_packets; + resolver.last_health_dns_responses = resolver.debug.dns_responses; + return; + } + if now.saturating_sub(resolver.last_health_check_at) < PATH_NO_PROGRESS_CHECK_US { + return; + } + + let send_delta = resolver + .debug + .send_packets + .saturating_sub(resolver.last_health_send_packets); + let response_delta = resolver + .debug + .dns_responses + .saturating_sub(resolver.last_health_dns_responses); + + resolver.last_health_check_at = now; + resolver.last_health_send_packets = resolver.debug.send_packets; + resolver.last_health_dns_responses = resolver.debug.dns_responses; + + if send_delta >= PATH_NO_PROGRESS_MIN_SENDS && response_delta == 0 { + warn!( + "Disabling resolver path {} for {}ms after {} outbound DNS packets with no responses", + resolver.addr, + PATH_NO_PROGRESS_DISABLE_US / 1000, + send_delta + ); + reset_resolver_path(resolver); + resolver.disabled_until = now.saturating_add(PATH_NO_PROGRESS_DISABLE_US); + } +} + +fn maybe_kick_recursive_polling( + resolver: &mut crate::dns::ResolverState, + now: u64, + active_streams: bool, +) { + if !active_streams + || !resolver.added + || resolver.mode != ResolverMode::Recursive + || resolver.pending_polls > 0 + { + return; + } + if resolver.last_active_poll_kick_at == 0 + || now.saturating_sub(resolver.last_active_poll_kick_at) >= RECURSIVE_ACTIVE_POLL_KICK_US + { + resolver.pending_polls = 1; + resolver.last_active_poll_kick_at = now; + } +} + pub async fn run_client(config: &ClientConfig<'_>) -> Result { let domain_len = config.domain.len(); let mtu = compute_mtu(domain_len)?; @@ -148,6 +219,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } unsafe { configure_quic_with_custom(quic, mixed_cc, mtu); + picoquic_set_default_idle_timeout( + quic, + config.quic_idle_timeout_seconds.saturating_mul(1000), + ); picoquic_enable_path_callbacks_default(quic, 1); let override_ptr = cc_override .as_ref() @@ -417,6 +492,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { 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() }; + let active_streams = streams_len > 0; if streams_len > 0 && has_ready_stream && flow_blocked { let now = unsafe { picoquic_current_time() }; if now.saturating_sub(last_flow_block_log_at) >= FLOW_BLOCKED_LOG_INTERVAL_US { @@ -453,6 +529,8 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if !refresh_resolver_path(cnx, resolver) { continue; } + let now = unsafe { picoquic_current_time() }; + maybe_kick_recursive_polling(resolver, now, active_streams); match resolver.mode { ResolverMode::Authoritative => { let quality = fetch_path_quality(cnx, resolver); @@ -540,7 +618,8 @@ 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() { + let active_streams = streams_len > 0; + for (idx, resolver) in resolvers.iter_mut().enumerate() { resolver.debug.enqueued_bytes = enqueued_bytes; resolver.debug.last_enqueue_at = last_enqueue_at; resolver.debug.zero_send_loops = zero_send_loops; @@ -571,6 +650,9 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { inflight_polls, resolver.last_pacing_snapshot, ); + if idx > 0 { + maybe_disable_no_progress_path(resolver, report_time, active_streams); + } } } diff --git a/crates/slipstream-client/src/streams/callback.rs b/crates/slipstream-client/src/streams/callback.rs index dd701a6d..25996846 100644 --- a/crates/slipstream-client/src/streams/callback.rs +++ b/crates/slipstream-client/src/streams/callback.rs @@ -23,6 +23,16 @@ fn close_event_label(event: picoquic_call_back_event_t) -> &'static str { } } +fn close_reason_label(reason: u64) -> &'static str { + match reason { + 0 => "none", + 0x101 => "slipstream_internal_error", + 0x105 => "slipstream_file_cancel_error", + 0x433 => "picoquic_idle_timeout", + _ => "unknown", + } +} + pub(crate) unsafe extern "C" fn client_callback( cnx: *mut picoquic_cnx_t, stream_id: u64, @@ -103,13 +113,17 @@ pub(crate) unsafe extern "C" fn client_callback( ); } warn!( - "Connection closed event={} state={:?} local_error=0x{:x} remote_error=0x{:x} local_app=0x{:x} remote_app=0x{:x} ready={}", + "Connection closed event={} state={:?} local_error=0x{:x}({}) remote_error=0x{:x}({}) local_app=0x{:x}({}) remote_app=0x{:x}({}) ready={}", close_event_label(fin_or_event), cnx_state, local_reason, + close_reason_label(local_reason), remote_reason, + close_reason_label(remote_reason), local_app_reason, + close_reason_label(local_app_reason), remote_app_reason, + close_reason_label(remote_app_reason), state.ready ); } diff --git a/crates/slipstream-ffi/src/lib.rs b/crates/slipstream-ffi/src/lib.rs index a54df7b5..48165172 100644 --- a/crates/slipstream-ffi/src/lib.rs +++ b/crates/slipstream-ffi/src/lib.rs @@ -32,6 +32,7 @@ pub struct ClientConfig<'a> { pub congestion_control: Option<&'a str>, pub gso: bool, pub keep_alive_interval: usize, + pub quic_idle_timeout_seconds: u64, pub debug_poll: bool, pub debug_streams: bool, } diff --git a/crates/slipstream-ffi/src/picoquic.rs b/crates/slipstream-ffi/src/picoquic.rs index 698c3e63..c143d8e3 100644 --- a/crates/slipstream-ffi/src/picoquic.rs +++ b/crates/slipstream-ffi/src/picoquic.rs @@ -203,6 +203,7 @@ extern "C" { pub fn picoquic_set_cookie_mode(quic: *mut picoquic_quic_t, cookie_mode: c_int); pub fn picoquic_set_default_priority(quic: *mut picoquic_quic_t, default_stream_priority: u8); + pub fn picoquic_set_default_idle_timeout(quic: *mut picoquic_quic_t, idle_timeout_ms: u64); pub fn picoquic_set_default_direct_receive_callback( quic: *mut picoquic_quic_t, direct_receive_fn: picoquic_stream_direct_receive_fn, diff --git a/crates/slipstream-server/src/main.rs b/crates/slipstream-server/src/main.rs index 144e62cf..e4c89443 100644 --- a/crates/slipstream-server/src/main.rs +++ b/crates/slipstream-server/src/main.rs @@ -43,6 +43,12 @@ struct Args { max_connections: u32, #[arg(long = "idle-timeout-seconds", default_value_t = 60)] idle_timeout_seconds: u64, + #[arg( + long = "keep-alive-interval", + default_value_t = 400, + help = "Send keep alive pings at this interval in milliseconds (disabled: 0)" + )] + keep_alive_interval: u64, #[arg(long = "debug-streams")] debug_streams: bool, #[arg(long = "debug-commands")] @@ -146,6 +152,15 @@ fn main() { } else { args.max_connections }; + let keep_alive_interval = if cli_provided(&matches, "keep_alive_interval") { + args.keep_alive_interval + } else if let Some(value) = + sip003::last_option_value(&sip003_env.plugin_options, "keep-alive-interval") + { + unwrap_or_exit(parse_keep_alive_interval(&value), "SIP003 env error", 2) + } else { + args.keep_alive_interval + }; let config = ServerConfig { dns_listen_host, @@ -158,6 +173,7 @@ fn main() { domains, max_connections, idle_timeout_seconds: args.idle_timeout_seconds, + keep_alive_interval_ms: keep_alive_interval, debug_streams: args.debug_streams, debug_commands: args.debug_commands, }; @@ -200,6 +216,13 @@ fn parse_max_connections(input: &str) -> Result { Ok(value) } +fn parse_keep_alive_interval(input: &str) -> Result { + let trimmed = input.trim(); + trimmed + .parse::() + .map_err(|_| format!("Invalid keep-alive-interval value: {}", trimmed)) +} + fn cli_provided(matches: &clap::ArgMatches, id: &str) -> bool { matches.value_source(id) == Some(ValueSource::CommandLine) } diff --git a/crates/slipstream-server/src/server.rs b/crates/slipstream-server/src/server.rs index 5b1f767f..36f47981 100644 --- a/crates/slipstream-server/src/server.rs +++ b/crates/slipstream-server/src/server.rs @@ -7,7 +7,8 @@ use slipstream_core::{ use slipstream_dns::{encode_response, Question, Rcode, ResponseParams}; use slipstream_ffi::picoquic::{ picoquic_cnx_t, picoquic_create, picoquic_current_time, picoquic_delete_cnx, - picoquic_get_first_cnx, picoquic_get_next_cnx, picoquic_prepare_packet_ex, picoquic_quic_t, + picoquic_enable_keep_alive, picoquic_get_first_cnx, picoquic_get_next_cnx, + picoquic_prepare_packet_ex, picoquic_quic_t, picoquic_set_default_idle_timeout, slipstream_has_ready_stream, slipstream_is_flow_blocked, slipstream_server_cc_algorithm, PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, }; @@ -81,6 +82,7 @@ pub struct ServerConfig { pub domains: Vec, pub max_connections: u32, pub idle_timeout_seconds: u64, + pub keep_alive_interval_ms: u64, pub debug_streams: bool, pub debug_commands: bool, } @@ -193,6 +195,7 @@ pub async fn run_server(config: &ServerConfig) -> Result { let debug_streams = config.debug_streams; let debug_commands = config.debug_commands; let idle_timeout = Duration::from_secs(config.idle_timeout_seconds); + let keep_alive_interval_us = config.keep_alive_interval_ms.saturating_mul(1000); let mut state = Box::new(ServerState::new( target_addr, command_tx, @@ -244,6 +247,7 @@ pub async fn run_server(config: &ServerConfig) -> Result { )); } configure_quic_with_custom(quic, slipstream_server_cc_algorithm, QUIC_MTU); + picoquic_set_default_idle_timeout(quic, config.idle_timeout_seconds.saturating_mul(1000)); } let udp = Arc::new(bind_udp_socket(&config.dns_listen_host, config.dns_listen_port).await?); @@ -280,6 +284,7 @@ pub async fn run_server(config: &ServerConfig) -> Result { let mut recv_buf = vec![0u8; recv_buf_len]; let mut send_buf = vec![0u8; PICOQUIC_MAX_PACKET_SIZE]; let mut last_seen = HashMap::new(); + let mut keep_alive_enabled = HashMap::new(); let mut last_idle_gc = Instant::now(); let mut last_flow_block_log_at: u64 = 0; @@ -362,6 +367,7 @@ pub async fn run_server(config: &ServerConfig) -> Result { quic, state_ptr, &mut last_seen, + &mut keep_alive_enabled, idle_timeout, &mut last_idle_gc, now, @@ -378,6 +384,16 @@ pub async fn run_server(config: &ServerConfig) -> Result { let loop_time = unsafe { picoquic_current_time() }; for slot in slots.iter_mut() { + if keep_alive_interval_us > 0 + && !slot.cnx.is_null() + && !keep_alive_enabled.contains_key(&(slot.cnx as usize)) + { + unsafe { + picoquic_enable_keep_alive(slot.cnx, keep_alive_interval_us); + } + keep_alive_enabled.insert(slot.cnx as usize, ()); + } + let mut send_length = 0usize; let mut addr_to: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; let mut addr_from: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; @@ -533,6 +549,7 @@ fn maybe_gc_idle_connections( quic: *mut picoquic_quic_t, state_ptr: *mut ServerState, last_seen: &mut HashMap, + keep_alive_enabled: &mut HashMap, idle_timeout: Duration, last_gc: &mut Instant, now: Instant, @@ -545,8 +562,10 @@ fn maybe_gc_idle_connections( } let active = collect_active_connections(quic); + keep_alive_enabled.retain(|cnx_id, _| active.contains_key(cnx_id)); if active.is_empty() { last_seen.clear(); + keep_alive_enabled.clear(); *last_gc = now; return; } @@ -578,6 +597,7 @@ fn maybe_gc_idle_connections( picoquic_delete_cnx(cnx); } last_seen.remove(&cnx_id); + keep_alive_enabled.remove(&cnx_id); } } *last_gc = now; diff --git a/crates/slipstream-server/tests/cert_pinning_e2e.rs b/crates/slipstream-server/tests/cert_pinning_e2e.rs index 435a8dbf..53a4d112 100644 --- a/crates/slipstream-server/tests/cert_pinning_e2e.rs +++ b/crates/slipstream-server/tests/cert_pinning_e2e.rs @@ -55,6 +55,7 @@ fn cert_pinning_e2e() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: false, diff --git a/crates/slipstream-server/tests/epipe_reset_e2e.rs b/crates/slipstream-server/tests/epipe_reset_e2e.rs index 8b0c576b..930625d8 100644 --- a/crates/slipstream-server/tests/epipe_reset_e2e.rs +++ b/crates/slipstream-server/tests/epipe_reset_e2e.rs @@ -123,6 +123,7 @@ fn epipe_triggers_quic_reset() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: true, @@ -204,14 +205,14 @@ fn epipe_triggers_quic_reset() { let saw_local_error = wait_for_any_log( &client_logs, &["tcp write error", "tcp read error"], - Duration::from_secs(2), + Duration::from_secs(10), ); if saw_local_error.is_none() { let snapshot = log_snapshot(&client_logs); panic!("expected client tcp read/write error\n{}", snapshot); } - if !wait_for_log(&server_logs, "reset event=", Duration::from_secs(2)) { + if !wait_for_log(&server_logs, "reset event=", Duration::from_secs(10)) { let client_snapshot = log_snapshot(&client_logs); let server_snapshot = log_snapshot(&server_logs); panic!( diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs index 3e16b64e..8c23ac03 100644 --- a/crates/slipstream-server/tests/flow_control_e2e.rs +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -142,6 +142,7 @@ fn setup_flow_control(envs: &[(&str, &str)]) -> Option { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs, rust_log: "info", capture_logs: true, diff --git a/crates/slipstream-server/tests/idle_gc_e2e.rs b/crates/slipstream-server/tests/idle_gc_e2e.rs index 422db661..5072689c 100644 --- a/crates/slipstream-server/tests/idle_gc_e2e.rs +++ b/crates/slipstream-server/tests/idle_gc_e2e.rs @@ -62,6 +62,7 @@ fn idle_gc_closes_connection() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: Some(1), + keep_alive_interval_ms: Some(0), envs: &[], rust_log: "debug", capture_logs: true, @@ -137,8 +138,14 @@ fn idle_gc_closes_connection() { let snapshot = log_snapshot(&client_logs); panic!("client did not accept TCP connection\n{}", snapshot); } - if !wait_for_log(&client_logs, "stateless_reset", Duration::from_secs(5)) { + if !wait_for_log(&client_logs, "stateless_reset", Duration::from_secs(5)) + && !wait_for_log( + &client_logs, + "picoquic_idle_timeout", + Duration::from_secs(1), + ) + { let snapshot = log_snapshot(&client_logs); - panic!("expected stateless reset close\n{}", snapshot); + panic!("expected idle close\n{}", snapshot); } } diff --git a/crates/slipstream-server/tests/restart_reconnect_e2e.rs b/crates/slipstream-server/tests/restart_reconnect_e2e.rs index aef27376..42ac1559 100644 --- a/crates/slipstream-server/tests/restart_reconnect_e2e.rs +++ b/crates/slipstream-server/tests/restart_reconnect_e2e.rs @@ -46,6 +46,7 @@ fn restart_reconnects_idle_client() { reset_seed_path: Some(&reset_seed_path), fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: false, @@ -93,6 +94,7 @@ fn restart_reconnects_idle_client() { reset_seed_path: Some(&reset_seed_path), fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: false, diff --git a/crates/slipstream-server/tests/stream_limit_e2e.rs b/crates/slipstream-server/tests/stream_limit_e2e.rs index 03caaa30..0f396b53 100644 --- a/crates/slipstream-server/tests/stream_limit_e2e.rs +++ b/crates/slipstream-server/tests/stream_limit_e2e.rs @@ -96,6 +96,7 @@ fn stream_limit_reuse_allows_next_stream() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: true, @@ -209,6 +210,7 @@ fn stream_limit_server_close_allows_next_stream() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: true, diff --git a/crates/slipstream-server/tests/support/mod.rs b/crates/slipstream-server/tests/support/mod.rs index 3ad4ac86..5e1f7beb 100644 --- a/crates/slipstream-server/tests/support/mod.rs +++ b/crates/slipstream-server/tests/support/mod.rs @@ -105,6 +105,7 @@ pub struct ServerArgs<'a> { pub reset_seed_path: Option<&'a Path>, pub fallback_addr: Option, pub idle_timeout_seconds: Option, + pub keep_alive_interval_ms: Option, pub envs: &'a [(&'a str, &'a str)], pub rust_log: &'a str, pub capture_logs: bool, @@ -198,6 +199,9 @@ pub fn spawn_server(args: ServerArgs<'_>) -> (ChildGuard, Option) { cmd.arg("--idle-timeout-seconds") .arg(idle_timeout.to_string()); } + if let Some(interval) = args.keep_alive_interval_ms { + cmd.arg("--keep-alive-interval").arg(interval.to_string()); + } for (key, value) in args.envs { cmd.env(key, value); } diff --git a/crates/slipstream-server/tests/target_read_error_reset_e2e.rs b/crates/slipstream-server/tests/target_read_error_reset_e2e.rs index 62e6b9fd..e6ad7aca 100644 --- a/crates/slipstream-server/tests/target_read_error_reset_e2e.rs +++ b/crates/slipstream-server/tests/target_read_error_reset_e2e.rs @@ -97,6 +97,7 @@ fn target_read_error_triggers_client_reset() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: true, diff --git a/crates/slipstream-server/tests/udp_fallback_e2e.rs b/crates/slipstream-server/tests/udp_fallback_e2e.rs index cbe8cb8b..46d73c73 100644 --- a/crates/slipstream-server/tests/udp_fallback_e2e.rs +++ b/crates/slipstream-server/tests/udp_fallback_e2e.rs @@ -122,6 +122,7 @@ fn udp_fallback_e2e() { reset_seed_path: None, fallback_addr: Some(echo.addr), idle_timeout_seconds: None, + keep_alive_interval_ms: None, envs: &[], rust_log: "info", capture_logs: false, diff --git a/docs/config.md b/docs/config.md index cb13bb0b..c9f633aa 100644 --- a/docs/config.md +++ b/docs/config.md @@ -47,8 +47,12 @@ certificates are not verified. - `--max-connections` Caps concurrent QUIC connections and sizes internal connection tables (default: 256). - `--idle-timeout-seconds` - Closes idle QUIC connections after the given number of seconds (default: 60). - Set to 0 to disable idle GC. + Sets picoquic's QUIC transport idle timeout and closes application-idle + connections after the given number of seconds (default: 60). Set to 0 to + disable both the transport idle timeout and idle GC. +- `--keep-alive-interval` + Sends server-initiated QUIC PINGs at the given millisecond interval + (default: 400). Set to 0 to disable server keep-alive. - `--reset-seed` Path to a 32-hex-char (16-byte) stateless reset seed. If the file does not exist, the server generates one and writes it with 0600 permissions. If not diff --git a/docs/sip003.md b/docs/sip003.md index b17c44e8..ff79fc1f 100644 --- a/docs/sip003.md +++ b/docs/sip003.md @@ -33,10 +33,11 @@ Supported keys: - `max-connections` - `congestion-control` - `keep-alive-interval` +- `quic-idle-timeout-seconds` -Client consumes `domain`, `resolver`, `authoritative`, `cert`, `congestion-control`, and -`keep-alive-interval`. Server consumes `domain`, `cert`, `key`, `reset-seed`, `fallback`, and -`max-connections`. +Client consumes `domain`, `resolver`, `authoritative`, `cert`, `congestion-control`, +`keep-alive-interval`, and `quic-idle-timeout-seconds`. Server consumes `domain`, `cert`, `key`, +`reset-seed`, `fallback`, `max-connections`, and `keep-alive-interval`. Syntax: `key=value;key=value`. Semicolons, equal signs, and backslashes must be escaped with backslashes (`\;`, `\=`, `\\`). @@ -45,7 +46,8 @@ List keys (`resolver`, `authoritative`) accept comma-separated values or repeate Order is preserved across keys, matching CLI ordering behavior. `domain` is a comma-separated list in a single option; repeating `domain` is invalid. The client requires exactly one domain value, while the server accepts multiple domains. -`keep-alive-interval` is in milliseconds (for example, `keep-alive-interval=400`). +`keep-alive-interval` is in milliseconds for both client and server (for example, `keep-alive-interval=400`). +`quic-idle-timeout-seconds` controls picoquic's transport idle timer. IPv6 resolver addresses must be bracketed, for example `[2001:db8::1]:53`. Using `authoritative` with an empty value (for example `authoritative=` or a bare `authoritative` entry) switches the `SS_REMOTE_*` fallback into authoritative mode. diff --git a/docs/usage.md b/docs/usage.md index aa7e3667..ae63baa2 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -20,6 +20,7 @@ Common flags: - --authoritative (repeatable; mark a resolver path as authoritative and use pacing-based polling) - --gso (currently not implemented in the Rust loop; prints a warning) - --keep-alive-interval (default: 400) +- --quic-idle-timeout-seconds (default: 120; set to 0 for no QUIC transport idle timeout) Example: @@ -46,6 +47,7 @@ Notes: - When --congestion-control is omitted, authoritative paths default to bbr and recursive paths default to dcubic. - Authoritative polling derives its QPS budget from picoquic’s pacing rate (scaled by the DNS payload size and RTT proxy) and falls back to cwnd if pacing is unavailable; `--debug-poll` logs the pacing rate, target QPS, and inflight polls. - When QUIC has ready stream data queued, authoritative polling yields to data-bearing queries unless flow control blocks progress. +- Keep-alive packets and the QUIC idle timeout are separate controls: `--keep-alive-interval` schedules PINGs, while `--quic-idle-timeout-seconds` controls picoquic's transport idle timer. - Expect higher CPU usage and detectability risk; misusing it can overload resolvers/servers. ## slipstream-server @@ -65,9 +67,12 @@ Common flags: - --target-address (default: 127.0.0.1:5201) - --max-connections (default: 256; caps concurrent QUIC connections) - --fallback (optional; forward non-DNS packets to this UDP endpoint) -- --idle-timeout-seconds (default: 60; set to 0 to disable) +- --idle-timeout-seconds (default: 60; set to 0 to disable transport idle timeout and idle GC) +- --keep-alive-interval (default: 400; set to 0 to disable server-initiated PINGs) - --reset-seed (optional; 32 hex chars / 16 bytes; auto-created if missing) - When binding the default `--dns-listen-host ::`, slipstream falls back to `0.0.0.0` if IPv6 is unavailable on the host. +- The server uses `--idle-timeout-seconds` for both picoquic's QUIC transport idle timeout and the application-level idle connection GC sweeper. +- The server sends QUIC keep-alive PINGs using `--keep-alive-interval`; keep this lower than the negotiated idle timeout on lossy DNS paths. - When binding to `::`, slipstream still attempts to enable dual-stack (IPV6_V6ONLY=0); if your OS disallows it, IPv4 DNS clients require sysctl changes or binding to an IPv4 address. - With --fallback enabled, peers that have recently sent DNS stay DNS-only; while active they switch to fallback only after 16 consecutive non-DNS packets to avoid diverting DNS on stray traffic. DNS-only classification expires after an idle timeout without DNS traffic. - Fallback sessions are created per source address without a hard cap; untrusted or spoofed UDP traffic can consume file descriptors/CPU. Use network filtering or rate limiting when exposing fallback to the public Internet, or disable --fallback if this is a concern. From c01f165b3e9cec9adc2b9e6d7bb4aa8c13933d88 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 11:36:35 +0300 Subject: [PATCH 02/19] Fix QUIC path failover after DNS path loss --- crates/slipstream-client/src/runtime.rs | 5 +- crates/slipstream-client/src/runtime/path.rs | 56 ++++++++++++++- crates/slipstream-ffi/cc/slipstream_poll.c | 49 +++++++++++++ crates/slipstream-ffi/src/picoquic.rs | 6 ++ crates/slipstream-server/src/server.rs | 76 +++++++++++--------- 5 files changed, 154 insertions(+), 38 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 1bd79ddc..ee8f8398 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -2,8 +2,8 @@ mod path; 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, + apply_path_mode, drain_path_events, ensure_default_path_available, fetch_path_quality, + find_resolver_by_addr_mut, loop_burst_total, path_poll_burst_max, }; use self::setup::{bind_tcp_listener, bind_udp_socket, compute_mtu, map_io}; use crate::dns::{ @@ -297,6 +297,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); } add_paths(cnx, &mut resolvers)?; + ensure_default_path_available(cnx, &mut resolvers, current_time)?; for resolver in resolvers.iter_mut() { if resolver.added { apply_path_mode(cnx, resolver)?; diff --git a/crates/slipstream-client/src/runtime/path.rs b/crates/slipstream-client/src/runtime/path.rs index c9343097..cf7e9ba2 100644 --- a/crates/slipstream-client/src/runtime/path.rs +++ b/crates/slipstream-client/src/runtime/path.rs @@ -6,11 +6,13 @@ use crate::error::ClientError; use crate::streams::{ClientState, PathEvent}; 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_promote_path_to_default, 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; @@ -87,6 +89,56 @@ pub(crate) fn drain_path_events( } } +pub(crate) fn ensure_default_path_available( + cnx: *mut picoquic_cnx_t, + resolvers: &mut [ResolverState], + current_time: u64, +) -> Result<(), ClientError> { + let Some((primary, alternates)) = resolvers.split_first_mut() else { + return Ok(()); + }; + if refresh_resolver_path(cnx, primary) { + return Ok(()); + } + let mut candidate = None; + for resolver in alternates.iter_mut() { + if resolver.added + && resolver + .unique_path_id + .is_some_and(|unique_path_id| unique_path_id != 0) + && refresh_resolver_path(cnx, resolver) + { + candidate = Some(resolver); + break; + } + } + let Some(candidate) = candidate else { + return Ok(()); + }; + let Some(unique_path_id) = candidate.unique_path_id else { + return Ok(()); + }; + if candidate.path_id == 0 { + return Ok(()); + } + let ret = unsafe { slipstream_promote_path_to_default(cnx, unique_path_id, current_time) }; + if ret != 0 { + warn!( + "Primary resolver path became unavailable; failed promoting {} to default path", + candidate.addr + ); + return Ok(()); + } + warn!( + "Primary resolver path became unavailable; promoted {} to default path", + candidate.addr + ); + for resolver in resolvers.iter_mut() { + refresh_resolver_path(cnx, resolver); + } + Ok(()) +} + 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) }; diff --git a/crates/slipstream-ffi/cc/slipstream_poll.c b/crates/slipstream-ffi/cc/slipstream_poll.c index ad2d4900..9570c94e 100644 --- a/crates/slipstream-ffi/cc/slipstream_poll.c +++ b/crates/slipstream-ffi/cc/slipstream_poll.c @@ -68,6 +68,55 @@ int slipstream_get_path_id_from_unique(picoquic_cnx_t *cnx, uint64_t unique_path return path_id; } +static int slipstream_is_path_id_usable(picoquic_cnx_t *cnx, int path_id) { + if (cnx == NULL || path_id < 0 || path_id >= cnx->nb_paths) { + return 0; + } + picoquic_path_t* path_x = cnx->path[path_id]; + if (path_x == NULL) { + return 0; + } + if (path_x->path_is_demoted || path_x->path_abandon_received || path_x->path_abandon_sent) { + return 0; + } + if (path_x->p_remote_cnxid == NULL || path_x->p_local_cnxid == NULL) { + return 0; + } + return 1; +} + +int slipstream_prepare_path_id(picoquic_cnx_t *cnx, int requested_path_id) { + if (slipstream_is_path_id_usable(cnx, requested_path_id)) { + return requested_path_id; + } + if (slipstream_is_path_id_usable(cnx, 0)) { + return 0; + } + return -1; +} + +int slipstream_promote_path_to_default(picoquic_cnx_t *cnx, uint64_t unique_path_id, uint64_t current_time) { + if (cnx == NULL || unique_path_id == 0) { + return -1; + } + int path_id = picoquic_get_path_id_from_unique(cnx, unique_path_id); + if (path_id == 0) { + return 0; + } + if (path_id < 0 || path_id >= cnx->nb_paths) { + return -1; + } + picoquic_path_t* path_x = cnx->path[path_id]; + if (path_x == NULL) { + return -1; + } + if (path_x->path_is_demoted || path_x->path_abandon_received || path_x->path_abandon_sent) { + return -1; + } + picoquic_promote_path_to_default(cnx, path_id, current_time); + return 0; +} + uint64_t slipstream_get_max_streams_bidir_remote(picoquic_cnx_t *cnx) { if (cnx == NULL || cnx->remote_parameters_received == 0) { return 0; diff --git a/crates/slipstream-ffi/src/picoquic.rs b/crates/slipstream-ffi/src/picoquic.rs index c143d8e3..bc76b863 100644 --- a/crates/slipstream-ffi/src/picoquic.rs +++ b/crates/slipstream-ffi/src/picoquic.rs @@ -321,6 +321,12 @@ extern "C" { cnx: *mut picoquic_cnx_t, unique_path_id: u64, ) -> c_int; + pub fn slipstream_prepare_path_id(cnx: *mut picoquic_cnx_t, requested_path_id: c_int) -> c_int; + pub fn slipstream_promote_path_to_default( + cnx: *mut picoquic_cnx_t, + unique_path_id: u64, + current_time: u64, + ) -> c_int; pub fn slipstream_get_max_streams_bidir_remote(cnx: *mut picoquic_cnx_t) -> u64; pub fn slipstream_set_cc_override(alg_name: *const c_char); pub fn slipstream_set_default_path_mode(mode: c_int); diff --git a/crates/slipstream-server/src/server.rs b/crates/slipstream-server/src/server.rs index 36f47981..67390281 100644 --- a/crates/slipstream-server/src/server.rs +++ b/crates/slipstream-server/src/server.rs @@ -9,8 +9,8 @@ use slipstream_ffi::picoquic::{ picoquic_cnx_t, picoquic_create, picoquic_current_time, picoquic_delete_cnx, picoquic_enable_keep_alive, picoquic_get_first_cnx, picoquic_get_next_cnx, picoquic_prepare_packet_ex, picoquic_quic_t, picoquic_set_default_idle_timeout, - slipstream_has_ready_stream, slipstream_is_flow_blocked, slipstream_server_cc_algorithm, - PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, + slipstream_has_ready_stream, slipstream_is_flow_blocked, slipstream_prepare_path_id, + slipstream_server_cc_algorithm, PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, }; use slipstream_ffi::{ configure_quic_with_custom, socket_addr_to_storage, take_crypto_errors, QuicGuard, @@ -400,38 +400,45 @@ pub async fn run_server(config: &ServerConfig) -> Result { let mut if_index: libc::c_int = 0; if slot.payload_override.is_none() && slot.rcode.is_none() && !slot.cnx.is_null() { - let ret = unsafe { - picoquic_prepare_packet_ex( - slot.cnx, - slot.path_id, - loop_time, - send_buf.as_mut_ptr(), - send_buf.len(), - &mut send_length, - &mut addr_to, - &mut addr_from, - &mut if_index, - std::ptr::null_mut(), - ) - }; - if ret < 0 { - return Err(ServerError::new("Failed to prepare QUIC packet")); - } + let path_id = unsafe { slipstream_prepare_path_id(slot.cnx, slot.path_id) }; + if path_id < 0 { + tracing::debug!( + "No usable QUIC path for response: requested_path_id={}", + slot.path_id + ); + } else { + let ret = unsafe { + picoquic_prepare_packet_ex( + slot.cnx, + path_id, + loop_time, + send_buf.as_mut_ptr(), + send_buf.len(), + &mut send_length, + &mut addr_to, + &mut addr_from, + &mut if_index, + std::ptr::null_mut(), + ) + }; + if ret < 0 { + return Err(ServerError::new("Failed to prepare QUIC packet")); + } - if send_length == 0 { - let cnx_id = slot.cnx as usize; - let metrics = unsafe { (&*state_ptr).stream_debug_metrics(cnx_id) }; - if metrics.streams_total > 0 - && metrics.has_send_backlog() - && loop_time.saturating_sub(last_flow_block_log_at) - >= FLOW_BLOCKED_LOG_INTERVAL_US - { - let flow_blocked = unsafe { slipstream_is_flow_blocked(slot.cnx) != 0 }; - let has_ready_stream = - unsafe { slipstream_has_ready_stream(slot.cnx) != 0 }; - let send_backlog = - unsafe { (&*state_ptr).stream_send_backlog_summaries(cnx_id, 8) }; - tracing::warn!( + if send_length == 0 { + let cnx_id = slot.cnx as usize; + let metrics = unsafe { (&*state_ptr).stream_debug_metrics(cnx_id) }; + if metrics.streams_total > 0 + && metrics.has_send_backlog() + && loop_time.saturating_sub(last_flow_block_log_at) + >= FLOW_BLOCKED_LOG_INTERVAL_US + { + let flow_blocked = unsafe { slipstream_is_flow_blocked(slot.cnx) != 0 }; + let has_ready_stream = + unsafe { slipstream_has_ready_stream(slot.cnx) != 0 }; + let send_backlog = + unsafe { (&*state_ptr).stream_send_backlog_summaries(cnx_id, 8) }; + tracing::warn!( "server connection stalled: cnx={} streams={} streams_with_write_tx={} streams_with_data_rx={} queued_bytes_total={} streams_with_pending_data={} pending_chunks_total={} pending_bytes_total={} streams_with_pending_fin={} streams_with_fin_enqueued={} streams_with_target_fin_pending={} streams_with_send_pending={} streams_with_send_stash={} send_stash_bytes_total={} streams_discarding={} streams_close_after_flush={} multi_stream={} flow_blocked={} has_ready_stream={} send_backlog={:?}", cnx_id, metrics.streams_total, @@ -454,7 +461,8 @@ pub async fn run_server(config: &ServerConfig) -> Result { has_ready_stream, send_backlog ); - last_flow_block_log_at = loop_time; + last_flow_block_log_at = loop_time; + } } } } From 53c0e860b97b4f2decedb6d81f370c406db365e4 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 12:10:33 +0300 Subject: [PATCH 03/19] Fix authoritative DNS polling under active streams --- crates/slipstream-client/src/dns/poll.rs | 15 ++++++++-- crates/slipstream-client/src/dns/resolver.rs | 30 +++++++++++++++++++ crates/slipstream-client/src/runtime.rs | 22 +++++++++----- crates/slipstream-client/src/runtime/setup.rs | 30 ++++++++++++++----- 4 files changed, 79 insertions(+), 18 deletions(-) diff --git a/crates/slipstream-client/src/dns/poll.rs b/crates/slipstream-client/src/dns/poll.rs index 98c07701..72f39566 100644 --- a/crates/slipstream-client/src/dns/poll.rs +++ b/crates/slipstream-client/src/dns/poll.rs @@ -10,6 +10,7 @@ use tokio::net::UdpSocket as TokioUdpSocket; use super::path::refresh_resolver_path; use super::resolver::{sockaddr_storage_to_socket_addr, PeerAddrMode, ResolverState}; +use tracing::warn; const AUTHORITATIVE_POLL_TIMEOUT_US: u64 = 5_000_000; @@ -87,8 +88,18 @@ pub(crate) async fn send_poll_queries( resolver.debug.polls_sent = resolver.debug.polls_sent.saturating_add(1); let poll_id = *dns_id; - let qname = build_qname(&send_buf[..send_length], config.domain) - .map_err(|err| ClientError::new(err.to_string()))?; + let qname = match build_qname(&send_buf[..send_length], config.domain) { + Ok(qname) => qname, + Err(err) if err.to_string().contains("payload too large") => { + warn!( + "Dropping oversized poll packet for DNS query transport: packet_len={} domain={}", + send_length, + config.domain + ); + continue; + } + Err(err) => return Err(ClientError::new(err.to_string())), + }; let params = QueryParams { id: poll_id, qname: &qname, diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index 9b11d68b..c3b38b21 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -51,6 +51,9 @@ pub(crate) struct ResolverState { pub(crate) pacing_budget: Option, pub(crate) last_pacing_snapshot: Option, pub(crate) debug: DebugMetrics, + pub(crate) is_primary: bool, + pub(crate) path_loss_count: u32, + pub(crate) last_path_loss_at: u64, } impl ResolverState { @@ -105,16 +108,43 @@ pub(crate) fn resolve_resolvers( }, last_pacing_snapshot: None, debug: DebugMetrics::new(debug_poll), + is_primary, + path_loss_count: 0, + last_path_loss_at: 0, }); } Ok(resolved) } +const PATH_LOSS_WINDOW_US: u64 = 10_000_000; +const PATH_LOSS_DISABLE_AFTER: u32 = 3; +const PATH_LOSS_DISABLE_US: u64 = 30_000_000; + pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { warn!( "Path for resolver {} became unavailable; resetting state", resolver.addr ); + let now = unsafe { slipstream_ffi::picoquic::picoquic_current_time() }; + if !resolver.is_primary { + if resolver.last_path_loss_at == 0 + || now.saturating_sub(resolver.last_path_loss_at) > PATH_LOSS_WINDOW_US + { + resolver.path_loss_count = 1; + } else { + resolver.path_loss_count = resolver.path_loss_count.saturating_add(1); + } + resolver.last_path_loss_at = now; + if resolver.path_loss_count >= PATH_LOSS_DISABLE_AFTER { + resolver.disabled_until = now.saturating_add(PATH_LOSS_DISABLE_US); + resolver.path_loss_count = 0; + warn!( + "Path for resolver {} is flapping; cooling down for {}ms", + resolver.addr, + PATH_LOSS_DISABLE_US / 1000 + ); + } + } resolver.added = false; resolver.path_id = -1; resolver.unique_path_id = None; diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index ee8f8398..d8655ed9 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -133,8 +133,7 @@ fn maybe_kick_recursive_polling( } pub async fn run_client(config: &ClientConfig<'_>) -> Result { - let domain_len = config.domain.len(); - let mtu = compute_mtu(domain_len)?; + let mtu = compute_mtu(config.domain)?; let udp = bind_udp_socket().await?; let udp_local_addr = udp.local_addr().map_err(map_io)?; let peer_addr_mode = PeerAddrMode::from_local_addr(udp_local_addr); @@ -464,8 +463,18 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } - let qname = build_qname(&send_buf[..send_length], config.domain) - .map_err(|err| ClientError::new(err.to_string()))?; + let qname = match build_qname(&send_buf[..send_length], config.domain) { + Ok(qname) => qname, + Err(err) if err.to_string().contains("payload too large") => { + warn!( + "Dropping oversized QUIC packet for DNS query transport: packet_len={} domain={}", + send_length, + config.domain + ); + continue; + } + Err(err) => return Err(ClientError::new(err.to_string())), + }; let params = QueryParams { id: dns_id, qname: &qname, @@ -541,10 +550,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { .unwrap_or_else(|| cwnd_target_polls(quality.cwin, mtu)); let inflight_packets = inflight_packet_estimate(quality.bytes_in_transit, mtu); - let mut poll_deficit = pacing_target.saturating_sub(inflight_packets); - if has_ready_stream && !flow_blocked { - poll_deficit = 0; - } + let poll_deficit = pacing_target.saturating_sub(inflight_packets); if poll_deficit > 0 && resolver.debug.enabled { debug!( "cc_state: {} cwnd={} in_transit={} rtt_us={} flow_blocked={} deficit={}", diff --git a/crates/slipstream-client/src/runtime/setup.rs b/crates/slipstream-client/src/runtime/setup.rs index 1840a769..41797f6c 100644 --- a/crates/slipstream-client/src/runtime/setup.rs +++ b/crates/slipstream-client/src/runtime/setup.rs @@ -2,21 +2,18 @@ use crate::error::ClientError; use slipstream_core::net::{ bind_first_resolved_with_ipv4_fallback, bind_tcp_listener_addr, bind_udp_socket_addr, }; +use slipstream_dns::max_payload_len_for_domain; use tokio::net::{TcpListener as TokioTcpListener, UdpSocket as TokioUdpSocket}; -pub(crate) fn compute_mtu(domain_len: usize) -> Result { - if domain_len >= 240 { - return Err(ClientError::new( - "Domain name is too long for DNS transport", - )); - } - let mtu = ((240.0 - domain_len as f64) / 1.6) as u32; +pub(crate) fn compute_mtu(domain: &str) -> Result { + let mtu = + max_payload_len_for_domain(domain).map_err(|err| ClientError::new(err.to_string()))?; if mtu == 0 { return Err(ClientError::new( "MTU computed to zero; check domain length", )); } - Ok(mtu) + Ok(mtu as u32) } pub(crate) async fn bind_udp_socket() -> Result { @@ -41,3 +38,20 @@ pub(crate) async fn bind_tcp_listener( pub(crate) fn map_io(err: std::io::Error) -> ClientError { ClientError::new(err.to_string()) } + +#[cfg(test)] +mod tests { + use super::compute_mtu; + use slipstream_dns::{build_qname, max_payload_len_for_domain}; + + #[test] + fn mtu_matches_dns_query_payload_capacity() { + let domain = "t.taskboards.org"; + let mtu = compute_mtu(domain).expect("mtu") as usize; + let max_payload = max_payload_len_for_domain(domain).expect("max payload"); + + assert_eq!(mtu, max_payload); + assert!(build_qname(&vec![0; mtu], domain).is_ok()); + assert!(build_qname(&vec![0; mtu + 1], domain).is_err()); + } +} From 9e50148e0449887dd4e82eba4cece4ea6ac5c91a Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 13:35:01 +0300 Subject: [PATCH 04/19] Recover recursive DNS sessions after path degradation --- crates/slipstream-client/src/runtime.rs | 135 ++++++++++++++++++++++-- 1 file changed, 125 insertions(+), 10 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index d8655ed9..4ac0e311 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -53,6 +53,8 @@ const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; +const PATH_POOR_RESPONSE_MIN_SENDS: u64 = 32; +const PATH_POOR_RESPONSE_RATIO_DIVISOR: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { @@ -70,21 +72,21 @@ fn maybe_disable_no_progress_path( resolver: &mut crate::dns::ResolverState, now: u64, active_streams: bool, -) { +) -> bool { if !active_streams || !resolver.added { resolver.last_health_check_at = now; resolver.last_health_send_packets = resolver.debug.send_packets; resolver.last_health_dns_responses = resolver.debug.dns_responses; - return; + return false; } if resolver.last_health_check_at == 0 { resolver.last_health_check_at = now; resolver.last_health_send_packets = resolver.debug.send_packets; resolver.last_health_dns_responses = resolver.debug.dns_responses; - return; + return false; } if now.saturating_sub(resolver.last_health_check_at) < PATH_NO_PROGRESS_CHECK_US { - return; + return false; } let send_delta = resolver @@ -100,16 +102,22 @@ fn maybe_disable_no_progress_path( resolver.last_health_send_packets = resolver.debug.send_packets; resolver.last_health_dns_responses = resolver.debug.dns_responses; - if send_delta >= PATH_NO_PROGRESS_MIN_SENDS && response_delta == 0 { + let no_responses = send_delta >= PATH_NO_PROGRESS_MIN_SENDS && response_delta == 0; + let poor_response_ratio = send_delta >= PATH_POOR_RESPONSE_MIN_SENDS + && response_delta.saturating_mul(PATH_POOR_RESPONSE_RATIO_DIVISOR) < send_delta; + if no_responses || poor_response_ratio { warn!( - "Disabling resolver path {} for {}ms after {} outbound DNS packets with no responses", + "Disabling resolver path {} for {}ms after {} outbound DNS packets and {} DNS responses", resolver.addr, PATH_NO_PROGRESS_DISABLE_US / 1000, - send_delta + send_delta, + response_delta ); reset_resolver_path(resolver); resolver.disabled_until = now.saturating_add(PATH_NO_PROGRESS_DISABLE_US); + return true; } + false } fn maybe_kick_recursive_polling( @@ -132,6 +140,51 @@ fn maybe_kick_recursive_polling( } } +fn primary_recursive_path_unavailable(resolvers: &[crate::dns::ResolverState]) -> bool { + resolvers + .first() + .is_some_and(|resolver| resolver.mode == ResolverMode::Recursive && !resolver.added) +} + +fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { + if resolvers.is_empty() { + return; + } + resolvers.rotate_left(start_index % resolvers.len()); + for (idx, resolver) in resolvers.iter_mut().enumerate() { + resolver.is_primary = idx == 0; + resolver.added = idx == 0; + resolver.path_id = if idx == 0 { 0 } else { -1 }; + resolver.unique_path_id = if idx == 0 { Some(0) } else { None }; + resolver.local_addr_storage = None; + resolver.pending_polls = 0; + resolver.inflight_poll_ids.clear(); + resolver.last_pacing_snapshot = None; + } +} + +fn next_recursive_start_index(resolvers: &[crate::dns::ResolverState]) -> usize { + resolvers + .iter() + .enumerate() + .skip(1) + .find(|(_, resolver)| { + resolver.mode == ResolverMode::Recursive + && resolver.added + && resolver.debug.dns_responses > 0 + }) + .map(|(idx, _)| idx) + .or_else(|| { + resolvers + .iter() + .enumerate() + .skip(1) + .find(|(_, resolver)| resolver.mode == ResolverMode::Recursive) + .map(|(idx, _)| idx) + }) + .unwrap_or(0) +} + pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mtu = compute_mtu(config.domain)?; let udp = bind_udp_socket().await?; @@ -171,6 +224,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let _state = state; let mut reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); + let mut resolver_start_index = 0usize; loop { let mut resolvers = @@ -178,6 +232,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if resolvers.is_empty() { return Err(ClientError::new("At least one resolver is required")); } + resolver_start_index %= resolvers.len(); + if resolver_start_index > 0 { + rotate_resolvers_for_start(&mut resolvers, resolver_start_index); + } let mut local_addr_storage = socket_addr_to_storage(udp_local_addr); @@ -277,6 +335,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut zero_send_loops = 0u64; let mut zero_send_with_streams = 0u64; let mut last_flow_block_log_at = 0u64; + let mut reconnect_when_idle = false; loop { let current_time = unsafe { picoquic_current_time() }; @@ -304,6 +363,29 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); + if ready && primary_recursive_path_unavailable(&resolvers) { + let streams_len = unsafe { (*state_ptr).streams_len() }; + let next_start_index = next_recursive_start_index(&resolvers); + if next_start_index > 0 { + resolver_start_index = + (resolver_start_index + next_start_index) % resolvers.len(); + } + if streams_len == 0 { + warn!( + "Primary recursive resolver path unavailable while idle; reconnecting with {} as primary", + resolvers[next_start_index].addr + ); + break; + } + if !reconnect_when_idle { + warn!( + "Primary recursive resolver path unavailable with {} active stream(s); reconnecting with {} as primary after they drain", + streams_len, + resolvers[next_start_index].addr + ); + } + reconnect_when_idle = true; + } for resolver in resolvers.iter_mut() { if resolver.mode == ResolverMode::Authoritative { @@ -400,6 +482,15 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { drain_commands(cnx, state_ptr, &mut command_rx); drain_stream_data(cnx, state_ptr); drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); + if reconnect_when_idle { + let streams_len = unsafe { (*state_ptr).streams_len() }; + if streams_len == 0 { + warn!( + "Active streams drained after recursive resolver path loss; reconnecting" + ); + break; + } + } for _ in 0..packet_loop_send_max { let current_time = unsafe { picoquic_current_time() }; @@ -626,7 +717,8 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let (enqueued_bytes, last_enqueue_at) = unsafe { (*state_ptr).debug_snapshot() }; let streams_len = unsafe { (*state_ptr).streams_len() }; let active_streams = streams_len > 0; - for (idx, resolver) in resolvers.iter_mut().enumerate() { + let mut reconnect_after_primary_health_loss = None; + for resolver in resolvers.iter_mut() { resolver.debug.enqueued_bytes = enqueued_bytes; resolver.debug.last_enqueue_at = last_enqueue_at; resolver.debug.zero_send_loops = zero_send_loops; @@ -657,9 +749,23 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { inflight_polls, resolver.last_pacing_snapshot, ); - if idx > 0 { - maybe_disable_no_progress_path(resolver, report_time, active_streams); + if maybe_disable_no_progress_path(resolver, report_time, active_streams) + && resolver.is_primary + { + reconnect_after_primary_health_loss = Some(resolver.addr); + } + } + if let Some(primary_addr) = reconnect_after_primary_health_loss { + let next_start_index = next_recursive_start_index(&resolvers); + if next_start_index > 0 { + resolver_start_index = + (resolver_start_index + next_start_index) % resolvers.len(); } + warn!( + "Primary recursive resolver {} degraded; reconnecting with {} as primary", + primary_addr, resolvers[next_start_index].addr + ); + break; } } @@ -667,6 +773,15 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { picoquic_close(cnx, 0); } + let was_ready = unsafe { (*state_ptr).is_ready() }; + if !was_ready && config.resolvers.len() > 1 { + resolver_start_index = (resolver_start_index + 1) % config.resolvers.len(); + warn!( + "Connection failed before ready; next reconnect will try resolver slot {} as primary", + resolver_start_index + ); + } + unsafe { (*state_ptr).reset_for_reconnect(); } From 9cc12b5d65ea4caae60676f346fb5a5d1ff2dd03 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 14:13:26 +0300 Subject: [PATCH 05/19] Refresh recursive DNS sessions after large transfers --- crates/slipstream-client/src/runtime.rs | 55 +++++++++++++++---- .../slipstream-client/src/streams/callback.rs | 13 ++++- crates/slipstream-client/src/streams/state.rs | 11 ++++ 3 files changed, 66 insertions(+), 13 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 4ac0e311..4b590b67 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -49,6 +49,8 @@ const DNS_WAKE_DELAY_MAX_US: i64 = 10_000_000; const DNS_POLL_SLICE_US: u64 = 50_000; const RECONNECT_SLEEP_MIN_MS: u64 = 250; const RECONNECT_SLEEP_MAX_MS: u64 = 5_000; +const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER: u32 = 2; +const RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES: u64 = 4 * 1024 * 1024; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; @@ -146,6 +148,12 @@ fn primary_recursive_path_unavailable(resolvers: &[crate::dns::ResolverState]) - .is_some_and(|resolver| resolver.mode == ResolverMode::Recursive && !resolver.added) } +fn uses_recursive_transport(resolvers: &[crate::dns::ResolverState]) -> bool { + resolvers + .iter() + .any(|resolver| resolver.mode == ResolverMode::Recursive) +} + fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -187,9 +195,6 @@ fn next_recursive_start_index(resolvers: &[crate::dns::ResolverState]) -> usize pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mtu = compute_mtu(config.domain)?; - let udp = bind_udp_socket().await?; - let udp_local_addr = udp.local_addr().map_err(map_io)?; - let peer_addr_mode = PeerAddrMode::from_local_addr(udp_local_addr); let (command_tx, mut command_rx) = mpsc::unbounded_channel(); let data_notify = Arc::new(Notify::new()); @@ -225,8 +230,12 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); let mut resolver_start_index = 0usize; + let mut failed_before_ready_reconnects = 0u32; loop { + let udp = bind_udp_socket().await?; + let udp_local_addr = udp.local_addr().map_err(map_io)?; + let peer_addr_mode = PeerAddrMode::from_local_addr(udp_local_addr); let mut resolvers = resolve_resolvers(config.resolvers, mtu, config.debug_poll, peer_addr_mode)?; if resolvers.is_empty() { @@ -336,6 +345,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut zero_send_with_streams = 0u64; let mut last_flow_block_log_at = 0u64; let mut reconnect_when_idle = false; + let mut connection_was_ready = false; loop { let current_time = unsafe { picoquic_current_time() }; @@ -348,9 +358,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let ready = unsafe { (*state_ptr).is_ready() }; if ready { + connection_was_ready = true; unsafe { (*state_ptr).update_acceptor_limit(cnx); } + failed_before_ready_reconnects = 0; if reconnect_delay != Duration::from_millis(RECONNECT_SLEEP_MIN_MS) { reconnect_delay = Duration::from_millis(RECONNECT_SLEEP_MIN_MS); } @@ -717,6 +729,19 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let (enqueued_bytes, last_enqueue_at) = unsafe { (*state_ptr).debug_snapshot() }; let streams_len = unsafe { (*state_ptr).streams_len() }; let active_streams = streams_len > 0; + let completed_stream_rx_bytes = unsafe { (*state_ptr).completed_stream_rx_bytes() }; + if !active_streams + && completed_stream_rx_bytes >= RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES + && uses_recursive_transport(&resolvers) + { + unsafe { + (*state_ptr).clear_completed_stream_rx_bytes(); + } + return Err(ClientError::new(format!( + "Completed {} bytes over recursive DNS transport; exiting for clean session refresh", + completed_stream_rx_bytes + ))); + } let mut reconnect_after_primary_health_loss = None; for resolver in resolvers.iter_mut() { resolver.debug.enqueued_bytes = enqueued_bytes; @@ -773,13 +798,23 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { picoquic_close(cnx, 0); } - let was_ready = unsafe { (*state_ptr).is_ready() }; - if !was_ready && config.resolvers.len() > 1 { - resolver_start_index = (resolver_start_index + 1) % config.resolvers.len(); - warn!( - "Connection failed before ready; next reconnect will try resolver slot {} as primary", - resolver_start_index - ); + if !connection_was_ready { + failed_before_ready_reconnects = failed_before_ready_reconnects.saturating_add(1); + if config.resolvers.len() > 1 { + resolver_start_index = (resolver_start_index + 1) % config.resolvers.len(); + warn!( + "Connection failed before ready; next reconnect will try resolver slot {} as primary", + resolver_start_index + ); + } else { + warn!("Connection failed before ready; no alternate resolver slot available"); + } + if failed_before_ready_reconnects >= RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER { + return Err(ClientError::new(format!( + "Connection failed before ready {} times; exiting for supervisor restart", + failed_before_ready_reconnects + ))); + } } unsafe { diff --git a/crates/slipstream-client/src/streams/callback.rs b/crates/slipstream-client/src/streams/callback.rs index 25996846..39c57351 100644 --- a/crates/slipstream-client/src/streams/callback.rs +++ b/crates/slipstream-client/src/streams/callback.rs @@ -260,10 +260,17 @@ pub(super) fn handle_stream_data( unsafe { abort_stream_bidi(cnx, stream_id, SLIPSTREAM_FILE_CANCEL_ERROR) }; state.streams.remove(&stream_id); } else if remove_stream { - if debug_streams { - debug!("stream {}: finished", stream_id); + if let Some(stream) = state.streams.remove(&stream_id) { + state.completed_stream_rx_bytes = state + .completed_stream_rx_bytes + .saturating_add(stream.flow.rx_bytes); + if debug_streams { + debug!( + "stream {}: finished rx_bytes={} tx_bytes={}", + stream_id, stream.flow.rx_bytes, stream.tx_bytes + ); + } } - state.streams.remove(&stream_id); } check_stream_invariants(state, stream_id, "handle_stream_data"); diff --git a/crates/slipstream-client/src/streams/state.rs b/crates/slipstream-client/src/streams/state.rs index f773dbbb..9739e517 100644 --- a/crates/slipstream-client/src/streams/state.rs +++ b/crates/slipstream-client/src/streams/state.rs @@ -24,6 +24,7 @@ pub(crate) struct ClientState { pub(super) acceptor: acceptor::ClientAcceptor, pub(super) debug_enqueued_bytes: u64, pub(super) debug_last_enqueue_at: u64, + pub(super) completed_stream_rx_bytes: u64, pub(super) acceptor_limit_logged: bool, } @@ -102,6 +103,7 @@ impl ClientState { acceptor, debug_enqueued_bytes: 0, debug_last_enqueue_at: 0, + completed_stream_rx_bytes: 0, acceptor_limit_logged: false, } } @@ -130,6 +132,14 @@ impl ClientState { (self.debug_enqueued_bytes, self.debug_last_enqueue_at) } + pub(crate) fn completed_stream_rx_bytes(&self) -> u64 { + self.completed_stream_rx_bytes + } + + pub(crate) fn clear_completed_stream_rx_bytes(&mut self) { + self.completed_stream_rx_bytes = 0; + } + pub(crate) fn stream_debug_metrics(&self) -> ClientStreamMetrics { let mut metrics = ClientStreamMetrics::default(); for stream in self.streams.values() { @@ -218,6 +228,7 @@ impl ClientState { self.acceptor.reset(); self.debug_enqueued_bytes = 0; self.debug_last_enqueue_at = 0; + self.completed_stream_rx_bytes = 0; self.acceptor_limit_logged = false; } } From c562792568f4cfbe57d03db379998dd2958ed21b Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 14:31:07 +0300 Subject: [PATCH 06/19] Refresh recursive DNS sessions in-process --- crates/slipstream-client/src/runtime.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 4b590b67..3b351670 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -734,13 +734,16 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { && completed_stream_rx_bytes >= RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES && uses_recursive_transport(&resolvers) { + let primary_addr = resolvers[0].addr; unsafe { (*state_ptr).clear_completed_stream_rx_bytes(); } - return Err(ClientError::new(format!( - "Completed {} bytes over recursive DNS transport; exiting for clean session refresh", - completed_stream_rx_bytes - ))); + warn!( + "Completed {} bytes over recursive DNS transport; reconnecting with {} as primary for clean session refresh", + completed_stream_rx_bytes, + primary_addr + ); + break; } let mut reconnect_after_primary_health_loss = None; for resolver in resolvers.iter_mut() { From 4609d57a51b9a034c889f6ae0a721903620af8fd Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 14:49:06 +0300 Subject: [PATCH 07/19] Add DNS query cache busting --- crates/slipstream-client/src/dns/poll.rs | 4 +-- crates/slipstream-client/src/runtime.rs | 8 ++++-- crates/slipstream-client/src/runtime/setup.rs | 15 ++++++----- crates/slipstream-dns/src/codec.rs | 19 +++++++++++++- crates/slipstream-dns/src/lib.rs | 14 +++++++++++ crates/slipstream-dns/tests/multi_domain.rs | 25 +++++++++++++++++-- 6 files changed, 72 insertions(+), 13 deletions(-) diff --git a/crates/slipstream-client/src/dns/poll.rs b/crates/slipstream-client/src/dns/poll.rs index 72f39566..db1e97fb 100644 --- a/crates/slipstream-client/src/dns/poll.rs +++ b/crates/slipstream-client/src/dns/poll.rs @@ -1,6 +1,6 @@ use crate::error::ClientError; use slipstream_core::net::is_transient_udp_error; -use slipstream_dns::{build_qname, encode_query, QueryParams, CLASS_IN, RR_TXT}; +use slipstream_dns::{build_qname_with_nonce, encode_query, QueryParams, CLASS_IN, RR_TXT}; use slipstream_ffi::picoquic::{ picoquic_cnx_t, picoquic_current_time, picoquic_prepare_packet_ex, slipstream_request_poll, }; @@ -88,7 +88,7 @@ pub(crate) async fn send_poll_queries( resolver.debug.polls_sent = resolver.debug.polls_sent.saturating_add(1); let poll_id = *dns_id; - let qname = match build_qname(&send_buf[..send_length], config.domain) { + let qname = match build_qname_with_nonce(&send_buf[..send_length], config.domain, poll_id) { Ok(qname) => qname, Err(err) if err.to_string().contains("payload too large") => { warn!( diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 3b351670..615e4ccd 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -19,7 +19,7 @@ use crate::streams::{ ClientState, Command, }; use slipstream_core::net::is_transient_udp_error; -use slipstream_dns::{build_qname, encode_query, QueryParams, CLASS_IN, RR_TXT}; +use slipstream_dns::{build_qname_with_nonce, encode_query, QueryParams, CLASS_IN, RR_TXT}; use slipstream_ffi::{ configure_quic_with_custom, picoquic::{ @@ -566,7 +566,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } - let qname = match build_qname(&send_buf[..send_length], config.domain) { + let qname = match build_qname_with_nonce( + &send_buf[..send_length], + config.domain, + dns_id, + ) { Ok(qname) => qname, Err(err) if err.to_string().contains("payload too large") => { warn!( diff --git a/crates/slipstream-client/src/runtime/setup.rs b/crates/slipstream-client/src/runtime/setup.rs index 41797f6c..bdcda4c9 100644 --- a/crates/slipstream-client/src/runtime/setup.rs +++ b/crates/slipstream-client/src/runtime/setup.rs @@ -2,12 +2,12 @@ use crate::error::ClientError; use slipstream_core::net::{ bind_first_resolved_with_ipv4_fallback, bind_tcp_listener_addr, bind_udp_socket_addr, }; -use slipstream_dns::max_payload_len_for_domain; +use slipstream_dns::max_payload_len_for_domain_with_nonce; use tokio::net::{TcpListener as TokioTcpListener, UdpSocket as TokioUdpSocket}; pub(crate) fn compute_mtu(domain: &str) -> Result { - let mtu = - max_payload_len_for_domain(domain).map_err(|err| ClientError::new(err.to_string()))?; + let mtu = max_payload_len_for_domain_with_nonce(domain) + .map_err(|err| ClientError::new(err.to_string()))?; if mtu == 0 { return Err(ClientError::new( "MTU computed to zero; check domain length", @@ -42,16 +42,19 @@ pub(crate) fn map_io(err: std::io::Error) -> ClientError { #[cfg(test)] mod tests { use super::compute_mtu; - use slipstream_dns::{build_qname, max_payload_len_for_domain}; + use slipstream_dns::{ + build_qname, build_qname_with_nonce, max_payload_len_for_domain_with_nonce, + }; #[test] fn mtu_matches_dns_query_payload_capacity() { let domain = "t.taskboards.org"; let mtu = compute_mtu(domain).expect("mtu") as usize; - let max_payload = max_payload_len_for_domain(domain).expect("max payload"); + let max_payload = max_payload_len_for_domain_with_nonce(domain).expect("max payload"); assert_eq!(mtu, max_payload); + assert!(build_qname_with_nonce(&vec![0; mtu], domain, 0x1234).is_ok()); + assert!(build_qname_with_nonce(&vec![0; mtu + 1], domain, 0x1234).is_err()); assert!(build_qname(&vec![0; mtu], domain).is_ok()); - assert!(build_qname(&vec![0; mtu + 1], domain).is_err()); } } diff --git a/crates/slipstream-dns/src/codec.rs b/crates/slipstream-dns/src/codec.rs index d89dbae4..0f4b439b 100644 --- a/crates/slipstream-dns/src/codec.rs +++ b/crates/slipstream-dns/src/codec.rs @@ -77,7 +77,8 @@ pub fn decode_query_with_domains( } }; - let undotted = dots::undotify(&subdomain_raw); + let subdomain_raw = strip_cache_buster_label(&subdomain_raw); + let undotted = dots::undotify(subdomain_raw); if undotted.is_empty() { return Err(DecodeQueryError::Reply { id: header.id, @@ -110,6 +111,22 @@ pub fn decode_query_with_domains( }) } +fn strip_cache_buster_label(subdomain: &str) -> &str { + let Some((payload, nonce)) = subdomain.rsplit_once('.') else { + return subdomain; + }; + if nonce.len() == 5 + && nonce.as_bytes()[0] == b'0' + && nonce.as_bytes()[1..] + .iter() + .all(|byte| byte.is_ascii_hexdigit()) + { + payload + } else { + subdomain + } +} + pub fn encode_query(params: &QueryParams<'_>) -> Result, DnsError> { let mut out = Vec::with_capacity(256); let mut flags = 0u16; diff --git a/crates/slipstream-dns/src/lib.rs b/crates/slipstream-dns/src/lib.rs index 950c4bbd..3e8f5d6f 100644 --- a/crates/slipstream-dns/src/lib.rs +++ b/crates/slipstream-dns/src/lib.rs @@ -30,6 +30,15 @@ pub fn build_qname(payload: &[u8], domain: &str) -> Result { Ok(format!("{}.{}.", dotted, domain)) } +pub fn build_qname_with_nonce( + payload: &[u8], + domain: &str, + nonce: u16, +) -> Result { + let domain = domain.trim_end_matches('.'); + build_qname(payload, &format!("0{nonce:04x}.{domain}")) +} + pub fn max_payload_len_for_domain(domain: &str) -> Result { let domain = domain.trim_end_matches('.'); if domain.is_empty() { @@ -59,6 +68,11 @@ pub fn max_payload_len_for_domain(domain: &str) -> Result { Ok(max_payload) } +pub fn max_payload_len_for_domain_with_nonce(domain: &str) -> Result { + let domain = domain.trim_end_matches('.'); + max_payload_len_for_domain(&format!("00000.{domain}")) +} + fn base32_len(payload_len: usize) -> usize { if payload_len == 0 { return 0; diff --git a/crates/slipstream-dns/tests/multi_domain.rs b/crates/slipstream-dns/tests/multi_domain.rs index dda5d13b..b9cf8067 100644 --- a/crates/slipstream-dns/tests/multi_domain.rs +++ b/crates/slipstream-dns/tests/multi_domain.rs @@ -1,6 +1,6 @@ use slipstream_dns::{ - build_qname, decode_query_with_domains, encode_query, DecodeQueryError, QueryParams, Rcode, - CLASS_IN, RR_TXT, + build_qname, build_qname_with_nonce, decode_query_with_domains, encode_query, DecodeQueryError, + QueryParams, Rcode, CLASS_IN, RR_TXT, }; #[test] @@ -45,6 +45,27 @@ fn decode_query_with_domains_prefers_longest_suffix() { assert_eq!(decoded.payload, payload); } +#[test] +fn decode_query_with_domains_strips_cache_buster_label() { + let payload = vec![5u8, 4, 3, 2, 1]; + let qname = build_qname_with_nonce(&payload, "example.com", 0x7abc).expect("build qname"); + let query = encode_query(&QueryParams { + id: 8, + qname: &qname, + qtype: RR_TXT, + qclass: CLASS_IN, + rd: true, + cd: false, + qdcount: 1, + is_query: true, + }) + .expect("encode query"); + + let decoded = + decode_query_with_domains(&query, &["example.com"]).expect("decode query with nonce"); + assert_eq!(decoded.payload, payload); +} + #[test] fn decode_query_with_domains_rejects_unknown_domain() { let payload = vec![1u8, 2, 3]; From 1a9b9f4206a876a9b4b16a7abfb3d95cd55e6bba Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 14:55:24 +0300 Subject: [PATCH 08/19] Avoid killing responsive recursive paths --- crates/slipstream-client/src/runtime.rs | 17 +++++------------ 1 file changed, 5 insertions(+), 12 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 615e4ccd..4cc473d2 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -55,8 +55,6 @@ const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; -const PATH_POOR_RESPONSE_MIN_SENDS: u64 = 32; -const PATH_POOR_RESPONSE_RATIO_DIVISOR: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { @@ -104,10 +102,7 @@ fn maybe_disable_no_progress_path( resolver.last_health_send_packets = resolver.debug.send_packets; resolver.last_health_dns_responses = resolver.debug.dns_responses; - let no_responses = send_delta >= PATH_NO_PROGRESS_MIN_SENDS && response_delta == 0; - let poor_response_ratio = send_delta >= PATH_POOR_RESPONSE_MIN_SENDS - && response_delta.saturating_mul(PATH_POOR_RESPONSE_RATIO_DIVISOR) < send_delta; - if no_responses || poor_response_ratio { + if send_delta >= PATH_NO_PROGRESS_MIN_SENDS && response_delta == 0 { warn!( "Disabling resolver path {} for {}ms after {} outbound DNS packets and {} DNS responses", resolver.addr, @@ -742,12 +737,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { unsafe { (*state_ptr).clear_completed_stream_rx_bytes(); } - warn!( - "Completed {} bytes over recursive DNS transport; reconnecting with {} as primary for clean session refresh", - completed_stream_rx_bytes, - primary_addr - ); - break; + return Err(ClientError::new(format!( + "Completed {} bytes over recursive DNS transport via {}; exiting for clean supervisor refresh", + completed_stream_rx_bytes, primary_addr + ))); } let mut reconnect_after_primary_health_loss = None; for resolver in resolvers.iter_mut() { From 5d9ba032886c8ddb009bda1bcecd57c40522ccc0 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 14:57:30 +0300 Subject: [PATCH 09/19] Rotate stalled handshakes before idle timeout --- crates/slipstream-client/src/runtime.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 4cc473d2..43ea47e7 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -50,6 +50,7 @@ const DNS_POLL_SLICE_US: u64 = 50_000; const RECONNECT_SLEEP_MIN_MS: u64 = 250; const RECONNECT_SLEEP_MAX_MS: u64 = 5_000; const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER: u32 = 2; +const RECONNECT_BEFORE_READY_TIMEOUT_US: u64 = 10_000_000; const RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES: u64 = 4 * 1024 * 1024; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; @@ -341,6 +342,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut last_flow_block_log_at = 0u64; let mut reconnect_when_idle = false; let mut connection_was_ready = false; + let connection_started_at = unsafe { picoquic_current_time() }; loop { let current_time = unsafe { picoquic_current_time() }; @@ -350,6 +352,16 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { if closing { break; } + if !connection_was_ready + && current_time.saturating_sub(connection_started_at) + >= RECONNECT_BEFORE_READY_TIMEOUT_US + { + warn!( + "Connection did not become ready within {}ms; reconnecting", + RECONNECT_BEFORE_READY_TIMEOUT_US / 1000 + ); + break; + } let ready = unsafe { (*state_ptr).is_ready() }; if ready { From 25f94acf317e65afbe358d06f8a899a01805f907 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:11:19 +0300 Subject: [PATCH 10/19] Keep recursive sessions alive after completed streams --- crates/slipstream-client/src/runtime.rs | 21 ------------------- .../slipstream-client/src/streams/callback.rs | 3 --- crates/slipstream-client/src/streams/state.rs | 11 ---------- 3 files changed, 35 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 43ea47e7..c35ecb52 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -51,7 +51,6 @@ const RECONNECT_SLEEP_MIN_MS: u64 = 250; const RECONNECT_SLEEP_MAX_MS: u64 = 5_000; const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER: u32 = 2; const RECONNECT_BEFORE_READY_TIMEOUT_US: u64 = 10_000_000; -const RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES: u64 = 4 * 1024 * 1024; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; @@ -144,12 +143,6 @@ fn primary_recursive_path_unavailable(resolvers: &[crate::dns::ResolverState]) - .is_some_and(|resolver| resolver.mode == ResolverMode::Recursive && !resolver.added) } -fn uses_recursive_transport(resolvers: &[crate::dns::ResolverState]) -> bool { - resolvers - .iter() - .any(|resolver| resolver.mode == ResolverMode::Recursive) -} - fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -740,20 +733,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let (enqueued_bytes, last_enqueue_at) = unsafe { (*state_ptr).debug_snapshot() }; let streams_len = unsafe { (*state_ptr).streams_len() }; let active_streams = streams_len > 0; - let completed_stream_rx_bytes = unsafe { (*state_ptr).completed_stream_rx_bytes() }; - if !active_streams - && completed_stream_rx_bytes >= RECURSIVE_REFRESH_AFTER_COMPLETED_RX_BYTES - && uses_recursive_transport(&resolvers) - { - let primary_addr = resolvers[0].addr; - unsafe { - (*state_ptr).clear_completed_stream_rx_bytes(); - } - return Err(ClientError::new(format!( - "Completed {} bytes over recursive DNS transport via {}; exiting for clean supervisor refresh", - completed_stream_rx_bytes, primary_addr - ))); - } let mut reconnect_after_primary_health_loss = None; for resolver in resolvers.iter_mut() { resolver.debug.enqueued_bytes = enqueued_bytes; diff --git a/crates/slipstream-client/src/streams/callback.rs b/crates/slipstream-client/src/streams/callback.rs index 39c57351..6a1ca01b 100644 --- a/crates/slipstream-client/src/streams/callback.rs +++ b/crates/slipstream-client/src/streams/callback.rs @@ -261,9 +261,6 @@ pub(super) fn handle_stream_data( state.streams.remove(&stream_id); } else if remove_stream { if let Some(stream) = state.streams.remove(&stream_id) { - state.completed_stream_rx_bytes = state - .completed_stream_rx_bytes - .saturating_add(stream.flow.rx_bytes); if debug_streams { debug!( "stream {}: finished rx_bytes={} tx_bytes={}", diff --git a/crates/slipstream-client/src/streams/state.rs b/crates/slipstream-client/src/streams/state.rs index 9739e517..f773dbbb 100644 --- a/crates/slipstream-client/src/streams/state.rs +++ b/crates/slipstream-client/src/streams/state.rs @@ -24,7 +24,6 @@ pub(crate) struct ClientState { pub(super) acceptor: acceptor::ClientAcceptor, pub(super) debug_enqueued_bytes: u64, pub(super) debug_last_enqueue_at: u64, - pub(super) completed_stream_rx_bytes: u64, pub(super) acceptor_limit_logged: bool, } @@ -103,7 +102,6 @@ impl ClientState { acceptor, debug_enqueued_bytes: 0, debug_last_enqueue_at: 0, - completed_stream_rx_bytes: 0, acceptor_limit_logged: false, } } @@ -132,14 +130,6 @@ impl ClientState { (self.debug_enqueued_bytes, self.debug_last_enqueue_at) } - pub(crate) fn completed_stream_rx_bytes(&self) -> u64 { - self.completed_stream_rx_bytes - } - - pub(crate) fn clear_completed_stream_rx_bytes(&mut self) { - self.completed_stream_rx_bytes = 0; - } - pub(crate) fn stream_debug_metrics(&self) -> ClientStreamMetrics { let mut metrics = ClientStreamMetrics::default(); for stream in self.streams.values() { @@ -228,7 +218,6 @@ impl ClientState { self.acceptor.reset(); self.debug_enqueued_bytes = 0; self.debug_last_enqueue_at = 0; - self.completed_stream_rx_bytes = 0; self.acceptor_limit_logged = false; } } From 8060b8bd7d09ffb017d829f4c4b50bcd2347555e Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:20:40 +0300 Subject: [PATCH 11/19] Keep recursive connections on alternate paths --- crates/slipstream-client/src/runtime.rs | 61 ++++++++++++------------- 1 file changed, 28 insertions(+), 33 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index c35ecb52..a8756e0f 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -137,10 +137,10 @@ fn maybe_kick_recursive_polling( } } -fn primary_recursive_path_unavailable(resolvers: &[crate::dns::ResolverState]) -> bool { +fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool { resolvers - .first() - .is_some_and(|resolver| resolver.mode == ResolverMode::Recursive && !resolver.added) + .iter() + .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { @@ -333,7 +333,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let mut zero_send_loops = 0u64; let mut zero_send_with_streams = 0u64; let mut last_flow_block_log_at = 0u64; - let mut reconnect_when_idle = false; let mut connection_was_ready = false; let connection_started_at = unsafe { picoquic_current_time() }; @@ -375,7 +374,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); - if ready && primary_recursive_path_unavailable(&resolvers) { + if ready && !has_available_recursive_path(&resolvers) { let streams_len = unsafe { (*state_ptr).streams_len() }; let next_start_index = next_recursive_start_index(&resolvers); if next_start_index > 0 { @@ -384,19 +383,17 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } if streams_len == 0 { warn!( - "Primary recursive resolver path unavailable while idle; reconnecting with {} as primary", + "No recursive resolver path available while idle; reconnecting with {} as primary", resolvers[next_start_index].addr ); break; } - if !reconnect_when_idle { - warn!( - "Primary recursive resolver path unavailable with {} active stream(s); reconnecting with {} as primary after they drain", - streams_len, - resolvers[next_start_index].addr - ); - } - reconnect_when_idle = true; + warn!( + "No recursive resolver path available with {} active stream(s); reconnecting with {} as primary", + streams_len, + resolvers[next_start_index].addr + ); + break; } for resolver in resolvers.iter_mut() { @@ -494,16 +491,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { drain_commands(cnx, state_ptr, &mut command_rx); drain_stream_data(cnx, state_ptr); drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); - if reconnect_when_idle { - let streams_len = unsafe { (*state_ptr).streams_len() }; - if streams_len == 0 { - warn!( - "Active streams drained after recursive resolver path loss; reconnecting" - ); - break; - } - } - for _ in 0..packet_loop_send_max { let current_time = unsafe { picoquic_current_time() }; let mut send_length: libc::size_t = 0; @@ -772,16 +759,24 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } if let Some(primary_addr) = reconnect_after_primary_health_loss { - let next_start_index = next_recursive_start_index(&resolvers); - if next_start_index > 0 { - resolver_start_index = - (resolver_start_index + next_start_index) % resolvers.len(); + if has_available_recursive_path(&resolvers) { + ensure_default_path_available(cnx, &mut resolvers, report_time)?; + warn!( + "Primary recursive resolver {} degraded; continuing on an alternate recursive path", + primary_addr + ); + } else { + let next_start_index = next_recursive_start_index(&resolvers); + if next_start_index > 0 { + resolver_start_index = + (resolver_start_index + next_start_index) % resolvers.len(); + } + warn!( + "Primary recursive resolver {} degraded and no alternate recursive path is available; reconnecting with {} as primary", + primary_addr, resolvers[next_start_index].addr + ); + break; } - warn!( - "Primary recursive resolver {} degraded; reconnecting with {} as primary", - primary_addr, resolvers[next_start_index].addr - ); - break; } } From 69934dd7c634900657226c02c525c307382aaeea Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:33:12 +0300 Subject: [PATCH 12/19] Honor cooldown when refreshing DNS paths --- crates/slipstream-client/src/dns/path.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index ff0b7979..1c01deb5 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -18,6 +18,17 @@ pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, resolver: &mut ResolverState, ) -> bool { + let now = unsafe { picoquic_current_time() }; + if resolver.disabled_until > now { + resolver.added = false; + resolver.path_id = -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; + return false; + } if let Some(unique_path_id) = resolver.unique_path_id { let path_id = unsafe { slipstream_get_path_id_from_unique(cnx, unique_path_id) }; if path_id >= 0 { From 47e367c3ba2e181dc05e3d62442f16d9272be86b Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:46:03 +0300 Subject: [PATCH 13/19] Require responsive alternate DNS paths --- crates/slipstream-client/src/runtime.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index a8756e0f..c60f39ff 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -143,6 +143,14 @@ fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } +fn has_responsive_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool { + resolvers.iter().any(|resolver| { + resolver.mode == ResolverMode::Recursive + && resolver.added + && resolver.debug.dns_responses > 0 + }) +} + fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -759,7 +767,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } if let Some(primary_addr) = reconnect_after_primary_health_loss { - if has_available_recursive_path(&resolvers) { + if has_responsive_recursive_path(&resolvers) { ensure_default_path_available(cnx, &mut resolvers, report_time)?; warn!( "Primary recursive resolver {} degraded; continuing on an alternate recursive path", From 4834a3fb8b666ba9a31e70cfec3fc359f1807aa5 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:49:45 +0300 Subject: [PATCH 14/19] Revert "Require responsive alternate DNS paths" This reverts commit 47e367c3ba2e181dc05e3d62442f16d9272be86b. --- crates/slipstream-client/src/runtime.rs | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index c60f39ff..a8756e0f 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -143,14 +143,6 @@ fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } -fn has_responsive_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool { - resolvers.iter().any(|resolver| { - resolver.mode == ResolverMode::Recursive - && resolver.added - && resolver.debug.dns_responses > 0 - }) -} - fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -767,7 +759,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } if let Some(primary_addr) = reconnect_after_primary_health_loss { - if has_responsive_recursive_path(&resolvers) { + if has_available_recursive_path(&resolvers) { ensure_default_path_available(cnx, &mut resolvers, report_time)?; warn!( "Primary recursive resolver {} degraded; continuing on an alternate recursive path", From 7437e34caf551e1c4b52f10afcd40277e71089c1 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:55:36 +0300 Subject: [PATCH 15/19] Replicate recursive DNS queries across paths --- crates/slipstream-client/src/dns/poll.rs | 13 +++++++ crates/slipstream-client/src/runtime.rs | 45 ++++++++++++++++++++++-- 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/crates/slipstream-client/src/dns/poll.rs b/crates/slipstream-client/src/dns/poll.rs index db1e97fb..7ee37826 100644 --- a/crates/slipstream-client/src/dns/poll.rs +++ b/crates/slipstream-client/src/dns/poll.rs @@ -6,6 +6,7 @@ use slipstream_ffi::picoquic::{ }; use slipstream_ffi::{ClientConfig, ResolverMode}; use std::collections::HashMap; +use std::net::SocketAddr; use tokio::net::UdpSocket as TokioUdpSocket; use super::path::refresh_resolver_path; @@ -41,6 +42,7 @@ pub(crate) async fn send_poll_queries( peer_addr_mode: PeerAddrMode, remaining: &mut usize, send_buf: &mut [u8], + replica_dests: &[SocketAddr], ) -> Result<(), ClientError> { if !refresh_resolver_path(cnx, resolver) { return Ok(()); @@ -123,6 +125,17 @@ pub(crate) async fn send_poll_queries( } return Err(ClientError::new(err.to_string())); } + for replica in replica_dests { + if *replica == dest { + continue; + } + if let Err(err) = udp.send_to(&packet, *replica).await { + if is_transient_udp_error(&err) { + continue; + } + return Err(ClientError::new(err.to_string())); + } + } if resolver.mode == ResolverMode::Authoritative { resolver.inflight_poll_ids.insert(poll_id, current_time); } diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index a8756e0f..0a33d982 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -36,6 +36,7 @@ use slipstream_ffi::{ socket_addr_to_storage, take_crypto_errors, ClientConfig, QuicGuard, ResolverMode, }; use std::ffi::CString; +use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::sync::{mpsc, Notify}; @@ -56,6 +57,7 @@ const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; +const RECURSIVE_REPLICA_LIMIT: usize = 2; fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { let mut dropped = 0usize; @@ -143,6 +145,28 @@ fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } +fn recursive_replica_dests( + resolvers: &[crate::dns::ResolverState], + primary_dest: SocketAddr, + now: u64, +) -> Vec { + let mut replicas = Vec::new(); + for resolver in resolvers { + if resolver.mode != ResolverMode::Recursive + || !resolver.added + || resolver.disabled_until > now + || resolver.addr == primary_dest + { + continue; + } + replicas.push(resolver.addr); + if replicas.len() >= RECURSIVE_REPLICA_LIMIT { + break; + } + } + replicas +} + fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -586,11 +610,19 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let dest = sockaddr_storage_to_socket_addr(&addr_to)?; let dest = peer_addr_mode.canonicalize(dest); local_addr_storage = addr_from; + let replica_dests = recursive_replica_dests(&resolvers, dest, current_time); if let Err(err) = udp.send_to(&packet, dest).await { if !is_transient_udp_error(&err) { return Err(map_io(err)); } } + for replica in replica_dests { + if let Err(err) = udp.send_to(&packet, replica).await { + if !is_transient_udp_error(&err) { + return Err(map_io(err)); + } + } + } } let has_ready_stream = unsafe { slipstream_has_ready_stream(cnx) != 0 }; @@ -629,11 +661,17 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { last_flow_block_log_at = now; } } - for resolver in resolvers.iter_mut() { - if !refresh_resolver_path(cnx, resolver) { + for idx in 0..resolvers.len() { + if !refresh_resolver_path(cnx, &mut resolvers[idx]) { continue; } let now = unsafe { picoquic_current_time() }; + let replica_dests = if resolvers[idx].mode == ResolverMode::Recursive { + recursive_replica_dests(&resolvers, resolvers[idx].addr, now) + } else { + Vec::new() + }; + let resolver = &mut resolvers[idx]; maybe_kick_recursive_polling(resolver, now, active_streams); match resolver.mode { ResolverMode::Authoritative => { @@ -669,6 +707,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut to_send, &mut send_buf, + &[], ) .await?; } @@ -689,6 +728,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut to_send, &mut send_buf, + &replica_dests, ) .await?; resolver.pending_polls = resolver @@ -707,6 +747,7 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut pending, &mut send_buf, + &replica_dests, ) .await?; resolver.pending_polls = pending; From e541c736d5ee43de72f794ad11755b959e69f7ab Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 15:58:41 +0300 Subject: [PATCH 16/19] Revert "Replicate recursive DNS queries across paths" This reverts commit 7437e34caf551e1c4b52f10afcd40277e71089c1. --- crates/slipstream-client/src/dns/poll.rs | 13 ------- crates/slipstream-client/src/runtime.rs | 45 ++---------------------- 2 files changed, 2 insertions(+), 56 deletions(-) diff --git a/crates/slipstream-client/src/dns/poll.rs b/crates/slipstream-client/src/dns/poll.rs index 7ee37826..db1e97fb 100644 --- a/crates/slipstream-client/src/dns/poll.rs +++ b/crates/slipstream-client/src/dns/poll.rs @@ -6,7 +6,6 @@ use slipstream_ffi::picoquic::{ }; use slipstream_ffi::{ClientConfig, ResolverMode}; use std::collections::HashMap; -use std::net::SocketAddr; use tokio::net::UdpSocket as TokioUdpSocket; use super::path::refresh_resolver_path; @@ -42,7 +41,6 @@ pub(crate) async fn send_poll_queries( peer_addr_mode: PeerAddrMode, remaining: &mut usize, send_buf: &mut [u8], - replica_dests: &[SocketAddr], ) -> Result<(), ClientError> { if !refresh_resolver_path(cnx, resolver) { return Ok(()); @@ -125,17 +123,6 @@ pub(crate) async fn send_poll_queries( } return Err(ClientError::new(err.to_string())); } - for replica in replica_dests { - if *replica == dest { - continue; - } - if let Err(err) = udp.send_to(&packet, *replica).await { - if is_transient_udp_error(&err) { - continue; - } - return Err(ClientError::new(err.to_string())); - } - } if resolver.mode == ResolverMode::Authoritative { resolver.inflight_poll_ids.insert(poll_id, current_time); } diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 0a33d982..a8756e0f 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -36,7 +36,6 @@ use slipstream_ffi::{ socket_addr_to_storage, take_crypto_errors, ClientConfig, QuicGuard, ResolverMode, }; use std::ffi::CString; -use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::sync::{mpsc, Notify}; @@ -57,7 +56,6 @@ const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; -const RECURSIVE_REPLICA_LIMIT: usize = 2; fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { let mut dropped = 0usize; @@ -145,28 +143,6 @@ fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } -fn recursive_replica_dests( - resolvers: &[crate::dns::ResolverState], - primary_dest: SocketAddr, - now: u64, -) -> Vec { - let mut replicas = Vec::new(); - for resolver in resolvers { - if resolver.mode != ResolverMode::Recursive - || !resolver.added - || resolver.disabled_until > now - || resolver.addr == primary_dest - { - continue; - } - replicas.push(resolver.addr); - if replicas.len() >= RECURSIVE_REPLICA_LIMIT { - break; - } - } - replicas -} - fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -610,19 +586,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { let dest = sockaddr_storage_to_socket_addr(&addr_to)?; let dest = peer_addr_mode.canonicalize(dest); local_addr_storage = addr_from; - let replica_dests = recursive_replica_dests(&resolvers, dest, current_time); if let Err(err) = udp.send_to(&packet, dest).await { if !is_transient_udp_error(&err) { return Err(map_io(err)); } } - for replica in replica_dests { - if let Err(err) = udp.send_to(&packet, replica).await { - if !is_transient_udp_error(&err) { - return Err(map_io(err)); - } - } - } } let has_ready_stream = unsafe { slipstream_has_ready_stream(cnx) != 0 }; @@ -661,17 +629,11 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { last_flow_block_log_at = now; } } - for idx in 0..resolvers.len() { - if !refresh_resolver_path(cnx, &mut resolvers[idx]) { + for resolver in resolvers.iter_mut() { + if !refresh_resolver_path(cnx, resolver) { continue; } let now = unsafe { picoquic_current_time() }; - let replica_dests = if resolvers[idx].mode == ResolverMode::Recursive { - recursive_replica_dests(&resolvers, resolvers[idx].addr, now) - } else { - Vec::new() - }; - let resolver = &mut resolvers[idx]; maybe_kick_recursive_polling(resolver, now, active_streams); match resolver.mode { ResolverMode::Authoritative => { @@ -707,7 +669,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut to_send, &mut send_buf, - &[], ) .await?; } @@ -728,7 +689,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut to_send, &mut send_buf, - &replica_dests, ) .await?; resolver.pending_polls = resolver @@ -747,7 +707,6 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { peer_addr_mode, &mut pending, &mut send_buf, - &replica_dests, ) .await?; resolver.pending_polls = pending; From 1463d2414c83beb30d16dca4200d01339c9495c5 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 16:10:26 +0300 Subject: [PATCH 17/19] Extend cooldown for failing DNS paths --- crates/slipstream-client/src/dns/path.rs | 2 +- crates/slipstream-client/src/dns/resolver.rs | 2 +- crates/slipstream-client/src/runtime.rs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 1c01deb5..7c1e3b76 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -12,7 +12,7 @@ use super::resolver::{reset_resolver_path, ResolverState}; const PATH_PROBE_INITIAL_DELAY_US: u64 = 250_000; const PATH_PROBE_MAX_DELAY_US: u64 = 10_000_000; const PATH_PROBE_DISABLE_AFTER_ATTEMPTS: u32 = 5; -const PATH_PROBE_DISABLE_US: u64 = 30_000_000; +const PATH_PROBE_DISABLE_US: u64 = 300_000_000; pub(crate) fn refresh_resolver_path( cnx: *mut picoquic_cnx_t, diff --git a/crates/slipstream-client/src/dns/resolver.rs b/crates/slipstream-client/src/dns/resolver.rs index c3b38b21..1a9654ff 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -118,7 +118,7 @@ pub(crate) fn resolve_resolvers( const PATH_LOSS_WINDOW_US: u64 = 10_000_000; const PATH_LOSS_DISABLE_AFTER: u32 = 3; -const PATH_LOSS_DISABLE_US: u64 = 30_000_000; +const PATH_LOSS_DISABLE_US: u64 = 300_000_000; pub(crate) fn reset_resolver_path(resolver: &mut ResolverState) { warn!( diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index a8756e0f..cdb5ac4c 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -53,7 +53,7 @@ const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER: u32 = 2; const RECONNECT_BEFORE_READY_TIMEOUT_US: u64 = 10_000_000; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; -const PATH_NO_PROGRESS_DISABLE_US: u64 = 30_000_000; +const PATH_NO_PROGRESS_DISABLE_US: u64 = 300_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; From 9cbe5d8074b629f4a40c13f5b9378eb6d8df60ec Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 18:36:25 +0300 Subject: [PATCH 18/19] Guard recursive failover for authoritative mode --- crates/slipstream-client/src/runtime.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index cdb5ac4c..0d0db272 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -143,6 +143,12 @@ fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool .any(|resolver| resolver.mode == ResolverMode::Recursive && resolver.added) } +fn has_recursive_resolver(resolvers: &[crate::dns::ResolverState]) -> bool { + resolvers + .iter() + .any(|resolver| resolver.mode == ResolverMode::Recursive) +} + fn rotate_resolvers_for_start(resolvers: &mut [crate::dns::ResolverState], start_index: usize) { if resolvers.is_empty() { return; @@ -374,7 +380,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); - if ready && !has_available_recursive_path(&resolvers) { + if ready + && has_recursive_resolver(&resolvers) + && !has_available_recursive_path(&resolvers) + { let streams_len = unsafe { (*state_ptr).streams_len() }; let next_start_index = next_recursive_start_index(&resolvers); if next_start_index > 0 { From 9eee7c99db643a05125065abeab443fe8d1e21c6 Mon Sep 17 00:00:00 2001 From: Evgenii Isaenkov Date: Thu, 14 May 2026 22:21:18 +0300 Subject: [PATCH 19/19] Scale pre-ready retries with resolver count --- crates/slipstream-client/src/runtime.rs | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/crates/slipstream-client/src/runtime.rs b/crates/slipstream-client/src/runtime.rs index 0d0db272..4d3a87fc 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -49,7 +49,7 @@ const DNS_WAKE_DELAY_MAX_US: i64 = 10_000_000; const DNS_POLL_SLICE_US: u64 = 50_000; const RECONNECT_SLEEP_MIN_MS: u64 = 250; const RECONNECT_SLEEP_MAX_MS: u64 = 5_000; -const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER: u32 = 2; +const RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER_MIN: u32 = 2; const RECONNECT_BEFORE_READY_TIMEOUT_US: u64 = 10_000_000; const FLOW_BLOCKED_LOG_INTERVAL_US: u64 = 1_000_000; const PATH_NO_PROGRESS_CHECK_US: u64 = 10_000_000; @@ -57,6 +57,10 @@ const PATH_NO_PROGRESS_DISABLE_US: u64 = 300_000_000; const PATH_NO_PROGRESS_MIN_SENDS: u64 = 8; const RECURSIVE_ACTIVE_POLL_KICK_US: u64 = 200_000; +fn reconnect_failed_before_ready_exit_after(resolver_count: usize) -> u32 { + resolver_count.max(RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER_MIN as usize) as u32 +} + fn drain_disconnected_commands(command_rx: &mut mpsc::UnboundedReceiver) -> usize { let mut dropped = 0usize; while let Ok(command) = command_rx.try_recv() { @@ -804,9 +808,10 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } else { warn!("Connection failed before ready; no alternate resolver slot available"); } - if failed_before_ready_reconnects >= RECONNECT_FAILED_BEFORE_READY_EXIT_AFTER { + let exit_after = reconnect_failed_before_ready_exit_after(resolvers.len()); + if failed_before_ready_reconnects >= exit_after { return Err(ClientError::new(format!( - "Connection failed before ready {} times; exiting for supervisor restart", + "Connection failed before ready {} times after trying configured resolver slots; exiting for supervisor restart", failed_before_ready_reconnects ))); } @@ -834,3 +839,16 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { reconnect_delay = (reconnect_delay * 2).min(Duration::from_millis(RECONNECT_SLEEP_MAX_MS)); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pre_ready_exit_after_covers_all_configured_resolvers() { + assert_eq!(reconnect_failed_before_ready_exit_after(0), 2); + assert_eq!(reconnect_failed_before_ready_exit_after(1), 2); + assert_eq!(reconnect_failed_before_ready_exit_after(2), 2); + assert_eq!(reconnect_failed_before_ready_exit_after(5), 5); + } +}