diff --git a/crates/slipstream-client/src/dns/path.rs b/crates/slipstream-client/src/dns/path.rs index 5042aa91..7c1e3b76 100644 --- a/crates/slipstream-client/src/dns/path.rs +++ b/crates/slipstream-client/src/dns/path.rs @@ -11,11 +11,24 @@ 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 = 300_000_000; 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 { @@ -64,6 +77,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 +106,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/poll.rs b/crates/slipstream-client/src/dns/poll.rs index 98c07701..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, }; @@ -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_with_nonce(&send_buf[..send_length], config.domain, poll_id) { + 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 320421e2..1a9654ff 100644 --- a/crates/slipstream-client/src/dns/resolver.rs +++ b/crates/slipstream-client/src/dns/resolver.rs @@ -41,11 +41,19 @@ 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, 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 { @@ -87,6 +95,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 { @@ -95,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 = 300_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; @@ -114,6 +154,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..4d3a87fc 100644 --- a/crates/slipstream-client/src/runtime.rs +++ b/crates/slipstream-client/src/runtime.rs @@ -2,14 +2,14 @@ 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::{ 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}; @@ -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::{ @@ -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, }; @@ -48,7 +49,17 @@ 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_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; +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; @@ -61,12 +72,128 @@ 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, +) -> 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 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 false; + } + if now.saturating_sub(resolver.last_health_check_at) < PATH_NO_PROGRESS_CHECK_US { + return false; + } + + 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 and {} DNS responses", + resolver.addr, + PATH_NO_PROGRESS_DISABLE_US / 1000, + 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( + 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; + } +} + +fn has_available_recursive_path(resolvers: &[crate::dns::ResolverState]) -> bool { + resolvers + .iter() + .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; + } + 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 domain_len = config.domain.len(); - let mtu = compute_mtu(domain_len)?; - 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 mtu = compute_mtu(config.domain)?; let (command_tx, mut command_rx) = mpsc::unbounded_channel(); let data_notify = Arc::new(Notify::new()); @@ -101,13 +228,22 @@ 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; + 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() { 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); @@ -148,6 +284,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() @@ -203,6 +343,8 @@ 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 connection_was_ready = false; + let connection_started_at = unsafe { picoquic_current_time() }; loop { let current_time = unsafe { picoquic_current_time() }; @@ -212,16 +354,29 @@ 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 { + 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); } 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)?; @@ -229,6 +384,30 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { } } drain_path_events(cnx, &mut resolvers, state_ptr, peer_addr_mode); + 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 { + resolver_start_index = + (resolver_start_index + next_start_index) % resolvers.len(); + } + if streams_len == 0 { + warn!( + "No recursive resolver path available while idle; reconnecting with {} as primary", + resolvers[next_start_index].addr + ); + break; + } + 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() { if resolver.mode == ResolverMode::Authoritative { @@ -325,7 +504,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); - for _ in 0..packet_loop_send_max { let current_time = unsafe { picoquic_current_time() }; let mut send_length: libc::size_t = 0; @@ -388,8 +566,22 @@ 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_with_nonce( + &send_buf[..send_length], + config.domain, + dns_id, + ) { + 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, @@ -417,6 +609,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 +646,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); @@ -462,10 +657,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={}", @@ -540,6 +732,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() }; + let active_streams = streams_len > 0; + 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; @@ -571,6 +765,31 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { inflight_polls, resolver.last_pacing_snapshot, ); + 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 { + 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; + } } } @@ -578,6 +797,26 @@ pub async fn run_client(config: &ClientConfig<'_>) -> Result { picoquic_close(cnx, 0); } + 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"); + } + 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 after trying configured resolver slots; exiting for supervisor restart", + failed_before_ready_reconnects + ))); + } + } + unsafe { (*state_ptr).reset_for_reconnect(); } @@ -600,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); + } +} 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-client/src/runtime/setup.rs b/crates/slipstream-client/src/runtime/setup.rs index 1840a769..bdcda4c9 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_with_nonce; 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_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", )); } - Ok(mtu) + Ok(mtu as u32) } pub(crate) async fn bind_udp_socket() -> Result { @@ -41,3 +38,23 @@ 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, 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_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()); + } +} diff --git a/crates/slipstream-client/src/streams/callback.rs b/crates/slipstream-client/src/streams/callback.rs index dd701a6d..6a1ca01b 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 ); } @@ -246,10 +260,14 @@ 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) { + 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-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]; 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/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..bc76b863 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, @@ -320,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/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..67390281 100644 --- a/crates/slipstream-server/src/server.rs +++ b/crates/slipstream-server/src/server.rs @@ -7,9 +7,10 @@ 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, - slipstream_has_ready_stream, slipstream_is_flow_blocked, slipstream_server_cc_algorithm, - PICOQUIC_MAX_PACKET_SIZE, PICOQUIC_PACKET_LOOP_RECV_MAX, + 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_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, @@ -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,44 +384,61 @@ 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() }; 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, @@ -438,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; + } } } } @@ -533,6 +557,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 +570,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 +605,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.