From 5fdbe696c57f2f0ed4a59d61deebeea1dfd7e2aa Mon Sep 17 00:00:00 2001 From: cfal Date: Sun, 25 Oct 2026 07:50:31 -0400 Subject: [PATCH] Make reload rollback safe and release TCP setup state (#219) * Release setup-only routing owners before TCP relay * Restore failed reloads and retain unchanged TCP listener sockets * Simplify reload activation and listener lifecycle code * Reject duplicate fixed TCP addresses during direct startup * Preserve inherited TUN descriptors across rejected reloads * Defer overlapping TCP listeners until retirement * Document reload rollback and listener retention * Exercise retained TCP sockets in lifecycle retention regression --- CHANGELOG.md | 1 + CONFIG.md | 29 +- src/certificates/runtime.rs | 14 + src/h2mux/h2mux_server_session.rs | 2 + src/http2/server.rs | 4 + src/hysteria2_server.rs | 2 + src/listener_tasks.rs | 21 +- src/naiveproxy/naive_hyper_service.rs | 2 + src/process.rs | 340 ++++++++++++++++--- src/tcp/tcp_server.rs | 333 +++++++++++++++++- src/tuic_server.rs | 2 + src/tun/mod.rs | 13 +- tests/cases/process_contract/certificates.rs | 20 ++ tests/cases/process_contract/clash_api.rs | 45 +++ tests/cases/process_contract/fake_dns.rs | 17 +- tests/cases/retention/process.rs | 4 + tests/process_contract.rs | 211 +++++++++++- 17 files changed, 987 insertions(+), 73 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0de97be8..ec68f7af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - Breaking: client groups now use `type` instead of `strategy`, with no legacy alias. Replace `strategy: selector` with `type: selector` and `strategy: round-robin` with `type: round-robin`; configurations containing `strategy` are rejected. - Added `type: url-test` client groups with automatic URL-scoped health/delay selection, tolerance, lazy checks, and supervised probes without a controller listener. See [upstream group configuration](CONFIG.md#clash-controller) and the [URL-test example](examples/url_test.yaml). +- Failed reload activation now restores the previous services instead of terminating, unless rollback also fails. Unchanged fixed-address TCP listeners retain their sockets, with new-generation TCP and Unix connections admitted only after activation completes. Existing TCP flows still obey the reload grace period after rollback; TUN, WireGuard, and QUIC remain stop/start. See [hot reloading](CONFIG.md#hot-reloading). - Breaking: file-backed inbound TLS/QUIC identities, including legacy `cert`/`key` paths and file-backed named PEMs, now rotate live and require currently valid, leaf-to-issuer certificate chains. Expired or misordered bundles can fail startup after upgrade, and expired sources stop serving new handshakes even to clients using fingerprint pinning or disabled verification. Renew certificates, remove expired cross-signs and certificates outside that chain, and synchronize the system clock before starting Shoes. Child-to-issuer signatures must use TLS-provider-supported algorithms (no SHA-1; RSA issuer keys of at least 2048 bits); reissue certificates with unsupported signatures. Inline-only pairs retain their previous behavior. See [certificate sources](CONFIG.md#certificate-sources). ## v0.3.4 diff --git a/CONFIG.md b/CONFIG.md index 14ab4961..086ba77a 100644 --- a/CONFIG.md +++ b/CONFIG.md @@ -248,11 +248,12 @@ throughput, and available memory rather than increasing buffers unconditionally. Periodic logs report endpoint flow/drop counts and each peer's public key, handshake age, and byte counters. Private keys and PSKs are redacted. -Preparation failures retain the old configuration. Successful reloads -disconnect existing tunnel flows and rebuild endpoints; sessions are not -migrated. Bind failures during activation can still terminate the process -after the old generation has stopped. Shutdown waits for buffered UDP tasks -to release their sockets before rebind. No host route cleanup is necessary. +Preparation failures retain the old configuration. Reload activation stops +existing tunnel flows and rebuilds endpoints; sessions are not migrated. +Bind failures during activation trigger rollback to the previous configuration, +but disconnected tunnel sessions are not restored. If rollback fails, Shoes +exits with status 1. Shutdown waits for buffered UDP tasks to release their +sockets before rebind. No host route cleanup is necessary. ## Configuration Structure @@ -1939,13 +1940,27 @@ reload loop: reload-only PEM changes require restarting the embedded instance. On Linux, if directory permissions prevent watching a readable configuration's parent, startup falls back to direct file watches and logs a warning. In-place edits remain detectable, but atomic replacements may be missed; use SIGHUP or grant directory read permission when replacing files. -Replacement configurations, certificates, and DNS resolvers are prepared while the current listeners continue serving. A preparation failure leaves the current configuration running. Once preparation succeeds, listeners are replaced. A failure to activate replacements exits with status 1; there is no rollback after listeners are retired. Existing TCP connections may drain for `global_limits.reload_grace_secs` during successful reloads; QUIC connections are closed. +Replacement configurations, certificates, and DNS resolvers are prepared while +the current listeners continue serving. A preparation failure leaves the current +configuration running. Unchanged fixed TCP listen addresses retain their sockets +across generations. New-generation TCP and Unix listeners accept queued +connections only after activation completes; connections may wait in the kernel +backlog during that interval. + +If activation fails, Shoes stops the candidate services and restores the previous +configuration, including control state, resource limits, DNS ownership, and +certificate workers. A successful rollback keeps the process running; if rollback +also fails, Shoes exits with status 1 rather than serving a partial generation. +Existing TCP connections begin draining when their old listeners retire, for up +to `global_limits.reload_grace_secs`. That deadline still applies after rollback. +TUN, WireGuard, and QUIC services are stopped and restarted; their existing +sessions are not migrated or restored by rollback. Certificate-source file changes and ACME renewal do not trigger this full reload. They rotate live without replacing listeners or closing QUIC connections; see [Certificate Sources](#certificate-sources). -QUIC listener sockets must finish closing before replacements bind. Closing endpoints get up to one second for best-effort close delivery, then their drivers are stopped locally so peer-controlled timers cannot delay reloads. A five-second socket-release guard remains; exceeding it indicates a local cleanup failure and exits with an error rather than announcing a partially ready generation. +QUIC listener sockets must finish closing before replacements bind. Closing endpoints get up to one second for best-effort close delivery, then their drivers are stopped locally so peer-controlled timers cannot delay reloads. If the previous generation's sockets do not retire within the five-second socket-release guard, Shoes exits with an error rather than announcing a partially ready generation. Candidate QUIC buffer sizes are used during DNS preparation, but `global_limits` admission caps change only at activation. DNS bootstrap over HTTP/3 or QUIC proxy chains therefore needs headroom under the live cap. A cap-increasing reload may require a restart if preparation cannot fit. Lowering a cap does not revoke existing reservations, including those made during preparation; new admissions are blocked until usage falls sufficiently. At cold startup, preparation uses the initial unlimited admission caps. diff --git a/src/certificates/runtime.rs b/src/certificates/runtime.rs index 2f7eb238..8338498a 100644 --- a/src/certificates/runtime.rs +++ b/src/certificates/runtime.rs @@ -213,6 +213,20 @@ impl Registry { Ok(()) } + pub(crate) fn listener_addresses(&self) -> Vec { + let mut addresses = Vec::new(); + for worker in &self.workers { + addresses.extend( + worker + .listeners + .lock() + .iter() + .filter_map(|listener| listener.local_addr().ok()), + ); + } + addresses + } + pub(crate) async fn start(&self) { let replacements = std::mem::take(&mut *self.replacements.lock()); for worker in replacements { diff --git a/src/h2mux/h2mux_server_session.rs b/src/h2mux/h2mux_server_session.rs index 71c53fe0..38b7b8ae 100644 --- a/src/h2mux/h2mux_server_session.rs +++ b/src/h2mux/h2mux_server_session.rs @@ -490,6 +490,8 @@ async fn handle_h2mux_tcp( let route = timeout_stream_setup(prepared.route(&mut stream, &proxy_selector, &resolver)).await?; let routed = timeout_stream_setup(route.connect(&resolver)).await?; + drop(proxy_selector); + drop(resolver); match routed.setup { Some(client_result) => { debug!( diff --git a/src/http2/server.rs b/src/http2/server.rs index cb7b1588..7ec44102 100644 --- a/src/http2/server.rs +++ b/src/http2/server.rs @@ -191,6 +191,8 @@ async fn tunnel( let Some(upstream) = routed.setup else { return Ok(()); }; + drop(selector); + drop(resolver); return relay(stream, upstream, routed.prefix).await; } let mut no_read = tokio::io::empty(); @@ -219,6 +221,8 @@ async fn tunnel( let send = respond .send_response(Response::new(()), false) .map_err(io::Error::other)?; + drop(selector); + drop(resolver); relay( H2MultiStream::new(send, request.into_body()), upstream, diff --git a/src/hysteria2_server.rs b/src/hysteria2_server.rs index 0140dffb..85731d2f 100644 --- a/src/hysteria2_server.rs +++ b/src/hysteria2_server.rs @@ -788,6 +788,8 @@ async fn relay_tcp_stream( Ok(routed.setup) } .await; + drop(client_proxy_selector); + drop(resolver); let Some(mut client_stream) = finish_tcp_setup(&mut server_stream, setup, success_sent) .await .map_err(|error| { diff --git a/src/listener_tasks.rs b/src/listener_tasks.rs index 863d2448..9e96459e 100644 --- a/src/listener_tasks.rs +++ b/src/listener_tasks.rs @@ -21,15 +21,18 @@ impl QuicRetirements { } pub async fn wait(&self) -> io::Result<()> { - let releases = std::mem::take(&mut *self.0.lock()); - tokio::time::timeout(Duration::from_secs(5), async { - for released in releases { + let mut releases = std::mem::take(&mut *self.0.lock()); + let result = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(released) = releases.last_mut() { // Sender destruction, rather than a sent value, confirms socket release. let _ = released.await; + releases.pop(); } }) - .await - .map_err(|_| { + .await; + // A rollback must still prove release after an activation wait times out. + self.0.lock().extend(releases); + result.map_err(|_| { io::Error::new( io::ErrorKind::TimedOut, "QUIC listener sockets did not retire", @@ -164,14 +167,18 @@ mod tests { use super::*; #[tokio::test(start_paused = true)] - async fn retirement_timeout_is_an_error() { + async fn activation_retirement_timeout_retains_pending_socket_releases() { let retirements = QuicRetirements::default(); - let (_socket, released) = oneshot::channel(); + let (socket, released) = oneshot::channel(); retirements.0.lock().push(released); assert_eq!( retirements.wait().await.unwrap_err().kind(), io::ErrorKind::TimedOut ); + assert_eq!(retirements.0.lock().len(), 1); + drop(socket); + retirements.wait().await.unwrap(); + assert!(retirements.0.lock().is_empty()); } #[tokio::test] diff --git a/src/naiveproxy/naive_hyper_service.rs b/src/naiveproxy/naive_hyper_service.rs index 009a8c24..fe378d3c 100644 --- a/src/naiveproxy/naive_hyper_service.rs +++ b/src/naiveproxy/naive_hyper_service.rs @@ -932,6 +932,8 @@ async fn handle_naive_stream( crate::util::timeout_stream_setup(prepared.route(&mut stream, &proxy_selector, &resolver)) .await?; let routed = crate::util::timeout_stream_setup(route.connect(&resolver)).await?; + drop(proxy_selector); + drop(resolver); let mut client_stream: Box = match routed.setup { Some(result) => { let flow = result.client_stream.flow().cloned(); diff --git a/src/process.rs b/src/process.rs index adbb46b1..55ca5ccb 100644 --- a/src/process.rs +++ b/src/process.rs @@ -12,6 +12,7 @@ use crate::dns::{self, DnsRegistry}; use crate::file_watcher::watch; use crate::option_util::OneOrSome; use crate::resources; +use crate::tcp::tcp_server::TcpListeners; const RELOAD_DEBOUNCE: Duration = Duration::from_secs(5); @@ -43,8 +44,11 @@ struct PreparedServers { certificates: crate::certificates::Registry, validated: ValidatedConfigs, dns: DnsRegistry, + tcp: TcpListeners, #[cfg(unix)] wireguard: crate::wireguard::Prepared, + #[cfg(unix)] + tun_fds: std::collections::HashMap>, } #[cfg(test)] @@ -182,13 +186,28 @@ async fn prepare_tracking( } else { None }; + #[cfg(unix)] + let tun_fds = validated + .configs + .iter() + .filter_map(|config| { + let Config::TunServer(tun) = config else { + return None; + }; + let fd = tun.device_fd?; + Some((fd, previous_servers?.tun_fds.get(&fd)?.clone())) + }) + .collect(); Ok(PreparedServers { generation, certificates, validated, dns, + tcp: TcpListeners::default(), #[cfg(unix)] wireguard, + #[cfg(unix)] + tun_fds, }) } @@ -279,7 +298,27 @@ impl RunningServers { let resolver = prepared.dns.get_for_server(dns_ref); #[cfg(unix)] if let Config::TunServer(config) = config { - match crate::tun::start_tun_server(config.clone(), resolver).await { + if let Some(fd) = config.device_fd { + use std::os::fd::FromRawFd; + if let std::collections::hash_map::Entry::Vacant(entry) = + prepared.tun_fds.entry(fd) + { + if unsafe { libc::fcntl(fd, libc::F_GETFD) } < 0 { + return Err(io::Error::last_os_error()); + } + entry.insert(std::sync::Arc::new(unsafe { + std::os::fd::OwnedFd::from_raw_fd(fd) + })); + } + } + // The registry retains the descriptor through rollback; each stack borrows a duplicate. + match crate::tun::start_tun_server_with_fd_ownership( + config.clone(), + resolver, + false, + ) + .await + { Ok(task) => self.tun.push(task), Err(error) => { self.stop().await; @@ -288,7 +327,11 @@ impl RunningServers { } continue; } - let startup = crate::tcp::tcp_server::start_servers(config.clone(), resolver); + let startup = crate::tcp::tcp_server::start_servers_with_listeners( + config.clone(), + resolver, + &mut prepared.tcp, + ); match self.quic_retirements.track(startup).await { Ok(tasks) => self.listeners.extend(tasks), Err(error) => { @@ -297,9 +340,73 @@ impl RunningServers { } } } - prepared.certificates.start().await; Ok(()) } + + async fn start_retry(&mut self, prepared: &mut PreparedServers) -> io::Result<()> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(3); + loop { + match self.start(prepared).await { + Ok(()) => return Ok(()), + Err(error) => { + self.stop().await; + if error.kind() != io::ErrorKind::AddrInUse + || tokio::time::Instant::now() >= deadline + { + return Err(error); + } + self.quic_retirements.wait().await?; + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + } + } + + async fn activate( + &mut self, + prepared: &mut PreparedServers, + control: Option<&crate::control::runtime::ControlService>, + ) -> io::Result<()> { + self.start_retry(prepared).await?; + if let Some(control) = control { + if let Some(generation) = &prepared.generation { + control.runtime.publish(generation.clone()).await?; + } + control.runtime.resume().await?; + } + // Workers can update durable ACME state, so activation starts them last. + prepared.certificates.start().await; + prepared.tcp.activate(); + Ok(()) + } + + async fn rollback( + &mut self, + previous: &mut PreparedServers, + candidate: &mut PreparedServers, + control: Option<&crate::control::runtime::ControlService>, + ) -> io::Result<()> { + self.stop().await; + candidate + .certificates + .retire_unused(&previous.certificates) + .await; + candidate.certificates.shutdown().await; + // Candidate-only sockets can overlap addresses the previous listeners rebind. + candidate.tcp = TcpListeners::default(); + // Rejected reloads retain inherited descriptors for a corrected configuration. + #[cfg(unix)] + for (fd, descriptor) in candidate.tun_fds.drain() { + previous.tun_fds.entry(fd).or_insert(descriptor); + } + self.quic_retirements.wait().await?; + // Pools may be shared with the live registry: rejection is not terminal shutdown. + if let Err(error) = candidate.dns.checkpoint().await { + log::warn!("Rejected candidate Fake DNS checkpoint failed: {error}"); + } + previous.tcp.defer_admission(); + self.activate(previous, control).await + } } impl Drop for RunningServers { @@ -502,6 +609,8 @@ async fn run_inner( _ = shutdown.recv() => return Ok(()), result = prepare_tracking(&paths, None, None, &mut discovered, control) => result?, }; + prepared.tcp = + TcpListeners::prepare(&prepared.validated.configs, &TcpListeners::default(), &[])?; watches.activated(&prepared.validated.reload_files); let mut running = RunningServers::default(); if let (Some(control), Some(generation)) = (control.as_ref(), prepared.generation.as_ref()) { @@ -522,6 +631,8 @@ async fn run_inner( if let Some(control) = control.as_ref() { control.runtime.resume().await?; } + prepared.certificates.start().await; + prepared.tcp.activate(); println!("Servers ready"); loop { let debounce = tokio::select! { @@ -570,69 +681,64 @@ async fn run_inner( continue; } }; + candidate.tcp = match TcpListeners::prepare( + &candidate.validated.configs, + &prepared.tcp, + &prepared.certificates.listener_addresses(), + ) { + Ok(listeners) => listeners, + Err(error) => { + eprintln!("Reload rejected; current servers retained: {error}"); + continue; + } + }; if let (Some(control), Some(generation)) = (control.as_ref(), candidate.generation.as_ref()) && let Err(error) = control.runtime.begin_activation(generation).await { eprintln!("Reload rejected; current servers retained: {error}"); continue; } - running.stop().await; - prepared - .certificates - .retire_unused(&candidate.certificates) - .await; - if let Err(error) = prepared.dns.checkpoint().await { - log::warn!("Reload Fake DNS checkpoint failed; continuing reload: {error}"); - } + let retirement = async { + running.stop().await; + prepared.tcp.retain_shared(&candidate.tcp); + prepared + .certificates + .retire_unused(&candidate.certificates) + .await; + if let Err(error) = prepared.dns.checkpoint().await { + log::warn!("Reload Fake DNS checkpoint failed; continuing reload: {error}"); + } + running.quic_retirements.wait().await + }; tokio::select! { biased; _ = shutdown.recv() => return running.shutdown(&[&prepared, &candidate]).await, - result = running.quic_retirements.wait() => if let Err(error) = result { + result = retirement => if let Err(error) = result { let _ = running.shutdown(&[&prepared, &candidate]).await; return Err(error); }, } - let deadline = tokio::time::Instant::now() + Duration::from_secs(3); - if let (Some(control), Some(generation)) = (control.as_ref(), candidate.generation.as_ref()) - { - control.runtime.publish(generation.clone()).await?; - } - loop { - let result = tokio::select! { + let result = tokio::select! { + biased; + _ = shutdown.recv() => return running.shutdown(&[&prepared, &candidate]).await, + result = running.activate(&mut candidate, control.as_ref()) => result, + }; + if let Err(error) = result { + let rollback = tokio::select! { biased; _ = shutdown.recv() => return running.shutdown(&[&prepared, &candidate]).await, - result = running.start(&mut candidate) => result, + result = running.rollback(&mut prepared, &mut candidate, control.as_ref()) => result, }; - match result { - Ok(()) => break, - Err(error) - if error.kind() == io::ErrorKind::AddrInUse - && tokio::time::Instant::now() < deadline => - { - tokio::select! { - biased; - _ = shutdown.recv() => return running.shutdown(&[&prepared, &candidate]).await, - result = running.quic_retirements.wait() => if let Err(error) = result { - let _ = running.shutdown(&[&prepared, &candidate]).await; - return Err(error); - }, - } - tokio::select! { - biased; - _ = shutdown.recv() => return running.shutdown(&[&prepared, &candidate]).await, - _ = tokio::time::sleep(Duration::from_millis(50)) => {}, - } - } - Err(error) => { - let _ = running.shutdown(&[&prepared, &candidate]).await; - return Err(error); - } + if let Err(rollback) = rollback { + let cleanup = running.shutdown(&[&prepared, &candidate]).await; + return Err(io::Error::other(format!( + "Reload activation failed: {error}; rollback failed: {rollback}; shutdown: {cleanup:?}" + ))); } + eprintln!("Reload rejected; previous servers restored: {error}"); + continue; } watches.activated(&candidate.validated.reload_files); - if let Some(control) = control.as_ref() { - control.runtime.resume().await?; - } let mut retired = std::mem::replace(&mut prepared, candidate); retired.certificates.shutdown().await; println!("Servers ready"); @@ -647,6 +753,146 @@ mod retention_tests; mod tests { use super::*; + #[tokio::test] + async fn unchanged_tcp_socket_queues_connections_between_generations() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + tokio::time::timeout(Duration::from_secs(5), async { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.yaml"); + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reservation.local_addr().unwrap(); + drop(reservation); + std::fs::write( + &path, + format!("- address: '{address}'\n protocol: {{type: socks}}\n"), + ) + .unwrap(); + let paths = [path.to_str().unwrap().to_owned()]; + let mut previous = prepare(&paths, None).await.unwrap(); + let mut running = RunningServers::default(); + running.start(&mut previous).await.unwrap(); + let mut next = prepare(&paths, Some(&previous.dns)).await.unwrap(); + next.tcp = TcpListeners::prepare(&next.validated.configs, &previous.tcp, &[]).unwrap(); + running.stop().await; + previous.tcp.retain_shared(&next.tcp); + let mut queued = tokio::net::TcpStream::connect(address).await.unwrap(); + queued.write_all(&[5, 1, 0]).await.unwrap(); + drop(previous); + assert!(std::net::TcpListener::bind(address).is_err()); + running.start(&mut next).await.unwrap(); + let mut response = [0; 2]; + assert!( + tokio::time::timeout(Duration::from_millis(50), queued.read_exact(&mut response)) + .await + .is_err() + ); + next.tcp.activate(); + queued.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [5, 0]); + drop(queued); + running.shutdown(&[&next]).await.unwrap(); + drop(next); + crate::socket_util::new_tcp_listener(address, 128, None).unwrap(); + }) + .await + .unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn rejected_or_cancelled_activation_does_not_admit_candidate_dns_flows() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + tokio::time::timeout(Duration::from_secs(10), async { + for shutdown in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.yaml"); + let paths = [path.to_str().unwrap().to_owned()]; + std::fs::write( + &path, + "- address: '127.0.0.1:0'\n protocol: {type: socks}\n", + ) + .unwrap(); + let mut previous = prepare(&paths, None).await.unwrap(); + let mut running = RunningServers::default(); + running.start(&mut previous).await.unwrap(); + + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reservation.local_addr().unwrap(); + drop(reservation); + let unix_path = directory.path().join("listener.sock"); + std::fs::write( + &path, + serde_json::json!([ + {"dns_group":"candidate", "dns_servers":"udp://127.0.0.1:9", "fake_ip":{ + "state_file":directory.path().join("state.json"), "max_entries":16 + }}, + {"address":address.to_string(), "protocol":{"type":"socks"}, + "dns":{"servers":"candidate"}}, + {"path":unix_path, "protocol":{"type":"socks"}, + "dns":{"servers":"candidate"}} + ]) + .to_string(), + ) + .unwrap(); + let mut candidate = prepare(&paths, Some(&previous.dns)).await.unwrap(); + let resolver = + std::sync::Arc::downgrade(&candidate.dns.get_by_name("candidate").unwrap()); + candidate.tcp = + TcpListeners::prepare(&candidate.validated.configs, &previous.tcp, &[]) + .unwrap(); + running.stop().await; + previous.tcp.retain_shared(&candidate.tcp); + running.start(&mut candidate).await.unwrap(); + let mut tcp = tokio::net::TcpStream::connect(address).await.unwrap(); + let mut unix = tokio::net::UnixStream::connect(&unix_path).await.unwrap(); + // Pipelined negotiation would open a lease-owning DNS stream without the gate. + let request = [5, 1, 0, 5, 1, 0, 1, 198, 18, 0, 1, 0, 53]; + tcp.write_all(&request).await.unwrap(); + unix.write_all(&request).await.unwrap(); + for stream in [ + &mut tcp as &mut dyn crate::async_stream::AsyncStream, + &mut unix, + ] { + assert!( + tokio::time::timeout(Duration::from_millis(50), stream.read_u8()) + .await + .is_err(), + "uncommitted listener admitted a flow" + ); + } + if shutdown { + running.shutdown(&[&previous, &candidate]).await.unwrap(); + } else { + // Covers failure after listener startup but before activation commits. + running + .rollback(&mut previous, &mut candidate, None) + .await + .unwrap(); + } + drop(candidate); + assert!(resolver.upgrade().is_none()); + + // Queued clients stay open while the same persistent pool is acquired again. + let mut next = prepare(&paths, Some(&previous.dns)).await.unwrap(); + running.stop().await; + next.tcp = + TcpListeners::prepare(&next.validated.configs, &previous.tcp, &[]).unwrap(); + running.start(&mut next).await.unwrap(); + let mut queued = tokio::net::UnixStream::connect(&unix_path).await.unwrap(); + queued.write_all(&[5, 1, 0]).await.unwrap(); + next.tcp.activate(); + let mut response = [0; 2]; + queued.read_exact(&mut response).await.unwrap(); + assert_eq!(response, [5, 0]); + drop((tcp, unix, queued)); + running.shutdown(&[&previous, &next]).await.unwrap(); + } + }) + .await + .unwrap(); + } + #[tokio::test] async fn activation_discards_candidate_identity_events_but_preserves_reload_events() { tokio::time::timeout(Duration::from_secs(5), async { diff --git a/src/tcp/tcp_server.rs b/src/tcp/tcp_server.rs index a6830a71..d7a21eed 100644 --- a/src/tcp/tcp_server.rs +++ b/src/tcp/tcp_server.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use std::net::IpAddr; -use std::sync::Arc; +use std::net::{IpAddr, SocketAddr}; +use std::sync::{Arc, Weak}; use std::time::Duration; use log::{debug, error}; @@ -25,7 +25,7 @@ use crate::tun::start_tun_server; use crate::util::write_all; async fn run_tcp_server( - listener: tokio::net::TcpListener, + listener: Arc, tcp_config: TcpConfig, resolver: Arc, server_handler: Arc, @@ -230,6 +230,8 @@ pub(crate) async fn process_stream_with_permit( )); } }; + drop(proxy_selector); + drop(resolver); let Some(TcpClientSetupResult { mut client_stream, early_data, @@ -350,6 +352,7 @@ pub(crate) async fn process_stream_with_permit( .await } TcpServerSetupResult::Session(session) => { + drop(resolver); session.await; Ok(()) } @@ -397,9 +400,126 @@ pub async fn run_udp_copy( copy_result } +#[cfg(any(test, target_os = "android", target_os = "ios", feature = "ffi"))] pub async fn start_servers( config: Config, resolver: Arc, +) -> std::io::Result>> { + start_servers_with_listeners(config, resolver, &mut TcpListeners::default()).await +} + +/// Socket ownership is independent of handlers and routing generations. +#[derive(Default)] +pub(crate) struct TcpListeners { + sockets: HashMap>, + // Automatic ports participate in overlap checks without extending socket ownership. + automatic_sockets: Vec>, + admission: Option, +} + +impl TcpListeners { + pub(crate) fn prepare( + configs: &[Config], + previous: &Self, + retiring_http: &[SocketAddr], + ) -> std::io::Result { + let mut next = Self::default(); + next.defer_admission(); + let mut seen = std::collections::HashSet::new(); + for config in configs { + let Config::Server(server) = config else { + continue; + }; + if server.transport != Transport::Tcp { + continue; + } + let BindLocation::Address(addresses) = &server.bind_location else { + continue; + }; + for address in addresses.iter() { + for address in address.to_socket_addrs()? { + if address.port() == 0 { + continue; + } + if !seen.insert(address) { + return Err(std::io::Error::new( + std::io::ErrorKind::AddrInUse, + format!("duplicate TCP listener at {address}"), + )); + } + if let Some(listener) = previous.sockets.get(&address) { + next.sockets.insert(address, listener.clone()); + continue; + } + let automatic_addresses = previous + .automatic_sockets + .iter() + .filter_map(Weak::upgrade) + .filter_map(|listener| listener.local_addr().ok()); + let overlaps_retiring_listener = previous + .sockets + .values() + .filter_map(|listener| listener.local_addr().ok()) + .chain(automatic_addresses) + .chain(retiring_http.iter().copied()) + .any(|bound| { + let bound_ip = bound.ip().to_canonical(); + let requested_ip = address.ip().to_canonical(); + bound.port() == address.port() + && (bound_ip == requested_ip + || bound_ip.is_unspecified() + || requested_ip.is_unspecified()) + }); + // SO_REUSEADDR may permit overlapping binds before the old socket retires. + if overlaps_retiring_listener { + continue; + } + next.bind(address)?; + } + } + } + Ok(next) + } + + pub(crate) fn retain_shared(&mut self, next: &Self) { + self.sockets.retain(|address, listener| { + next.sockets + .get(address) + .is_some_and(|next| Arc::ptr_eq(listener, next)) + }); + } + + /// Gates subsequently started listeners without affecting already draining flows. + pub(crate) fn defer_admission(&mut self) { + self.admission = Some(tokio_util::sync::CancellationToken::new()); + } + + pub(crate) fn activate(&self) { + if let Some(admission) = &self.admission { + admission.cancel(); + } + } + + fn bind(&mut self, address: SocketAddr) -> std::io::Result> { + if let Some(listener) = self.sockets.get(&address) { + return Ok(listener.clone()); + } + let listener = Arc::new(new_tcp_listener(address, 4096, None)?); + if address.port() != 0 { + self.sockets.insert(address, listener.clone()); + } else { + self.automatic_sockets + .retain(|listener| listener.strong_count() > 0); + self.automatic_sockets.push(Arc::downgrade(&listener)); + } + Ok(listener) + } +} + +pub(crate) async fn start_servers_with_listeners( + config: Config, + resolver: Arc, + listeners: &mut TcpListeners, ) -> std::io::Result>> { match config { #[cfg(unix)] @@ -421,7 +541,9 @@ pub async fn start_servers( std::io::ErrorKind::Unsupported, "TUN server is not supported on this platform", )), - Config::Server(server_config) => start_tcp_or_quic_servers(server_config, resolver).await, + Config::Server(server_config) => { + start_tcp_or_quic_servers(server_config, resolver, listeners).await + } _ => Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, "Expected a server configuration", @@ -432,10 +554,11 @@ pub async fn start_servers( async fn start_tcp_or_quic_servers( config: ServerConfig, resolver: Arc, + listeners: &mut TcpListeners, ) -> std::io::Result>> { let bind_location = config.bind_location.to_string(); let join_handles = match config.transport { - Transport::Tcp => start_tcp_servers(config, resolver).await?, + Transport::Tcp => start_tcp_servers(config, resolver, listeners).await?, Transport::Quic => start_quic_servers(config, resolver).await?, Transport::Udp => { return Err(std::io::Error::new( @@ -458,6 +581,7 @@ async fn start_tcp_or_quic_servers( async fn start_tcp_servers( config: ServerConfig, resolver: Arc, + sockets: &mut TcpListeners, ) -> std::io::Result>> { let ServerConfig { bind_location, @@ -491,9 +615,16 @@ async fn start_tcp_servers( let mut handlers: HashMap> = HashMap::new(); let mut services = super::tcp_server_handler_factory::HandlerServices::default(); let mut listeners = Vec::new(); + let mut seen = std::collections::HashSet::new(); for address in addresses.into_vec() { for socket_addr in address.to_socket_addrs()? { - let listener = new_tcp_listener(socket_addr, 4096, None)?; + if socket_addr.port() != 0 && !seen.insert(socket_addr) { + return Err(std::io::Error::new( + std::io::ErrorKind::AddrInUse, + format!("duplicate TCP listener at {socket_addr}"), + )); + } + let listener = sockets.bind(socket_addr)?; let tcp_handler = match handlers.entry(socket_addr.ip()) { std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(), std::collections::hash_map::Entry::Vacant(entry) => entry.insert( @@ -514,7 +645,12 @@ async fn start_tcp_servers( for (listener, tcp_handler) in listeners { let tcp_config = tcp_config.clone(); let resolver = resolver.clone(); + let admission = sockets.admission.clone(); handles.push(tokio::spawn(async move { + // Uncommitted flows must not outlive rollback through graceful draining. + if let Some(admission) = admission { + admission.cancelled().await; + } if let Err(error) = run_tcp_server(listener, tcp_config, resolver, tcp_handler).await { @@ -537,7 +673,11 @@ async fn start_tcp_servers( let tcp_handler: Arc = create_tcp_server_handler(protocol, &client_proxy_selector, &resolver, None)? .into(); + let admission = sockets.admission.clone(); let handle = tokio::spawn(async move { + if let Some(admission) = admission { + admission.cancelled().await; + } if let Err(error) = run_unix_server(listener, resolver, tcp_handler).await { error!("Unix listener stopped: {error}"); } @@ -576,6 +716,100 @@ mod lifetime_tests { use crate::client_proxy_selector::{ClientProxySelector, ConnectAction, ConnectRule}; use crate::tcp::socket_connector::SocketConnector; + #[tokio::test] + async fn ephemeral_tcp_activation_does_not_cache_automatic_ports() { + let mut sockets = TcpListeners::default(); + let first = sockets.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let second = sockets.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + assert_ne!(first.local_addr().unwrap(), second.local_addr().unwrap()); + assert!(sockets.sockets.is_empty()); + assert_eq!(sockets.automatic_sockets.len(), 2); + drop((first, second)); + let _next = sockets.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + assert_eq!(sockets.automatic_sockets.len(), 1); + } + + #[tokio::test] + async fn tcp_preparation_defers_reported_http_overlap_before_binding() { + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reservation.local_addr().unwrap(); + let config = format!("- address: '{address}'\n protocol: {{type: http}}\n"); + let configs = + crate::config::create_server_configs(serde_yaml::from_str(&config).unwrap()).unwrap(); + drop(reservation); + for retiring in [ + address, + SocketAddr::new(std::net::Ipv4Addr::UNSPECIFIED.into(), address.port()), + ] { + let next = + TcpListeners::prepare(&configs.configs, &TcpListeners::default(), &[retiring]) + .unwrap(); + assert!(next.sockets.is_empty(), "overlapping listener was prebound"); + std::net::TcpListener::bind(address).expect("preparation claimed a retiring address"); + } + } + + #[tokio::test] + async fn tcp_preparation_defers_live_automatic_ports_without_retaining_them() { + let mut previous = TcpListeners::default(); + let live = previous.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let address = live.local_addr().unwrap(); + let config = format!("- address: '{address}'\n protocol: {{type: http}}\n"); + let configs = + crate::config::create_server_configs(serde_yaml::from_str(&config).unwrap()).unwrap(); + let next = TcpListeners::prepare(&configs.configs, &previous, &[]).unwrap(); + assert!( + next.sockets.is_empty(), + "automatic port was prebound or reused" + ); + drop(live); + let next = TcpListeners::prepare(&configs.configs, &previous, &[]).unwrap(); + assert!( + next.sockets.contains_key(&address), + "retired port was still deferred" + ); + } + + #[cfg(target_os = "linux")] + #[tokio::test] + async fn tcp_preparation_defers_ipv4_mapped_listener_transitions() { + use tokio::io::AsyncReadExt; + timeout(Duration::from_secs(5), async { + for (old_ip, new_ip) in [ + ("127.0.0.1", "::ffff:127.0.0.1"), + ("::ffff:127.0.0.1", "127.0.0.1"), + ] { + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + let mut previous = TcpListeners::default(); + let live = previous + .bind(SocketAddr::new(old_ip.parse().unwrap(), port)) + .unwrap(); + let address = SocketAddr::new(new_ip.parse().unwrap(), port); + let config = format!("- address: '{address}'\n protocol: {{type: http}}\n"); + let configs = + crate::config::create_server_configs(serde_yaml::from_str(&config).unwrap()) + .unwrap(); + let mut next = TcpListeners::prepare(&configs.configs, &previous, &[]).unwrap(); + assert!(next.sockets.is_empty(), "mapped address was prebound"); + drop(live); + previous.retain_shared(&next); + let replacement = next.bind(address).unwrap(); + let mut peer = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .unwrap(); + let (mut accepted, _) = replacement.accept().await.unwrap(); + peer.write_all(b"live").await.unwrap(); + let mut bytes = [0; 4]; + accepted.read_exact(&mut bytes).await.unwrap(); + assert_eq!(&bytes, b"live"); + } + }) + .await + .unwrap(); + } + #[tokio::test] async fn fake_dns_udp_modes() { use crate::routing::ingress::tests::{PacketMode, UdpCase, UdpFixture}; @@ -831,6 +1065,60 @@ mod lifetime_tests { assert!(registry.snapshot(0).connections.is_empty()); } + #[tokio::test] + async fn fixed_tcp_releases_setup_owners_before_relay_and_half_close() { + use tokio::io::AsyncReadExt; + timeout(Duration::from_secs(5), async { + let (inbound, mut peer) = tokio::io::duplex(64); + let (outbound, mut target) = tokio::io::duplex(64); + let chain = ClientProxyChain::new( + vec![InitialHopEntry::Direct(Arc::new(ReadyConnector( + std::sync::Mutex::new(Some(FlushStream { + inner: outbound, + stalled: false, + flushes: Arc::new(AtomicUsize::new(0)), + })), + )))], + vec![], + ); + let selector = Arc::new(ClientProxySelector::new(vec![ConnectRule::new( + vec![crate::address::NetLocationMask::ANY], + ConnectAction::new_allow(None, ClientChainGroup::new(vec![chain])), + )])); + let resolver: Arc = Arc::new(PendingResolver); + let handler = Arc::new(Handler(selector.clone())); + let weak_selector = Arc::downgrade(&selector); + let weak_resolver = Arc::downgrade(&resolver); + let weak_handler = Arc::downgrade(&handler); + drop(selector); + let relay = tokio::spawn(process_stream(inbound, handler, resolver)); + let mut response = [0; 8]; + peer.read_exact(&mut response).await.unwrap(); + assert_eq!(&response, b"response"); + let mut request = [0; 7]; + target.read_exact(&mut request).await.unwrap(); + assert_eq!(&request, b"request"); + assert!(weak_selector.upgrade().is_none()); + assert!(weak_resolver.upgrade().is_none()); + assert!(weak_handler.upgrade().is_none()); + assert!(!relay.is_finished()); + + peer.write_all(b"live request").await.unwrap(); + peer.shutdown().await.unwrap(); + let mut bytes = Vec::new(); + target.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"live request"); + target.write_all(b"reply after FIN").await.unwrap(); + target.shutdown().await.unwrap(); + bytes.clear(); + peer.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"reply after FIN"); + relay.await.unwrap().unwrap(); + }) + .await + .unwrap(); + } + #[tokio::test(start_paused = true)] async fn initial_writes_and_flushes_have_deadlines() { use tokio::io::AsyncReadExt; @@ -1202,6 +1490,39 @@ mod lifetime_tests { } } + #[tokio::test] + async fn direct_tcp_startup_rejects_duplicate_fixed_but_allows_automatic_ports() { + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reservation.local_addr().unwrap(); + drop(reservation); + for port in [address.port(), 0] { + let address = SocketAddr::new(address.ip(), port); + let config = + format!("- address: ['{address}', '{address}']\n protocol: {{type: http}}\n"); + let configs = + crate::config::create_server_configs(serde_yaml::from_str(&config).unwrap()) + .unwrap(); + let result = start_servers( + configs.configs.into_iter().next().unwrap(), + Arc::new(crate::resolver::NativeResolver::new()), + ) + .await; + if port == 0 { + let handles = result.unwrap(); + assert_eq!(handles.len(), 2); + for handle in &handles { + handle.abort(); + } + for handle in handles { + assert!(handle.await.unwrap_err().is_cancelled()); + } + } else { + assert_eq!(result.unwrap_err().kind(), io::ErrorKind::AddrInUse); + std::net::TcpListener::bind(address).expect("duplicate bind leaked a listener"); + } + } + } + #[tokio::test] async fn tcp_bind_failure_releases_all_prepared_listeners() { let first = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/src/tuic_server.rs b/src/tuic_server.rs index 28f915d5..6ca31731 100644 --- a/src/tuic_server.rs +++ b/src/tuic_server.rs @@ -473,6 +473,8 @@ async fn relay_tcp_stream( )); } }; + drop(client_proxy_selector); + drop(resolver); let Some(crate::tcp::tcp_handler::TcpClientSetupResult { mut client_stream, early_data, diff --git a/src/tun/mod.rs b/src/tun/mod.rs index c6883c54..41ddd852 100644 --- a/src/tun/mod.rs +++ b/src/tun/mod.rs @@ -303,6 +303,8 @@ pub(crate) async fn handle_tcp_connection( .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "TUN TCP setup timed out")) .and_then(|result| result) .inspect_err(|error| warn!("Failed to connect to {}: {}", target, error))?; + drop(proxy_selector); + drop(resolver); let Some((mut remote, wire_destination)) = connected else { debug!("TCP connection blocked by rules"); @@ -355,6 +357,14 @@ async fn handle_udp_packets( pub async fn start_tun_server( config: TunConfig, resolver: std::sync::Arc, +) -> std::io::Result>> { + start_tun_server_with_fd_ownership(config, resolver, true).await +} + +pub(crate) async fn start_tun_server_with_fd_ownership( + config: TunConfig, + resolver: std::sync::Arc, + close_fd_on_drop: bool, ) -> std::io::Result>> { let (shutdown_tx, shutdown_rx) = oneshot::channel(); let (ready_tx, ready_rx) = oneshot::channel(); @@ -363,7 +373,8 @@ pub async fn start_tun_server( let _keep_alive = shutdown_tx; let mut ready = Some(ready_tx); let result = - run_tun_from_config_inner(config, shutdown_rx, true, resolver, &mut ready).await; + run_tun_from_config_inner(config, shutdown_rx, close_fd_on_drop, resolver, &mut ready) + .await; if let Some(ready) = ready.take() { let _ = ready.send(result); Ok(()) diff --git a/tests/cases/process_contract/certificates.rs b/tests/cases/process_contract/certificates.rs index 7c8289f9..3aeffe3e 100644 --- a/tests/cases/process_contract/certificates.rs +++ b/tests/cases/process_contract/certificates.rs @@ -307,6 +307,16 @@ async fn reload_releases_retired_http01_ports_before_starting_proxy_listeners() child.wait_for("Servers ready").await; config[0]["challenge"] = json!({"type": "tls_alpn01"}); config.as_array_mut().unwrap().push(json!({"address": challenge.to_string(), "protocol": {"type": "socks"}})); + config.as_array_mut().unwrap().push(json!({"path":directory.path().join("missing/listener.sock"), "protocol":{"type":"socks"}})); + std::fs::write(&path, serde_yaml::to_string(&config).unwrap()).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Reload rejected; previous servers restored").await; + let mut restored = TcpStream::connect(challenge).await.unwrap(); + restored.write_all(b"GET / HTTP/1.1\r\nHost: proxy.test\r\n\r\n").await.unwrap(); + let mut response = String::new(); + restored.read_to_string(&mut response).await.unwrap(); + assert!(response.starts_with("HTTP/1.1 404 "), "{response}"); + config.as_array_mut().unwrap().pop(); std::fs::write(&path, serde_yaml::to_string(&config).unwrap()).unwrap(); child.signal(libc::SIGHUP); child.wait_for("Servers ready").await; @@ -330,6 +340,16 @@ async fn reload_releases_retired_http01_ports_before_starting_proxy_listeners() drop(stream); config[0]["challenge"]["listen"] = json!([challenge.to_string()]); + config.as_array_mut().unwrap().push(json!({"path":directory.path().join("missing/listener.sock"), "protocol":{"type":"socks"}})); + std::fs::write(&path, serde_yaml::to_string(&config).unwrap()).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Reload rejected; previous servers restored").await; + let mut stream = TcpStream::connect(challenge).await.unwrap(); + stream.write_all(&[5, 1, 0]).await.unwrap(); + stream.read_exact(&mut reply).await.unwrap(); + assert_eq!(reply, [5, 0], "failed HTTP-01 activation must restore the proxy"); + drop(stream); + config.as_array_mut().unwrap().pop(); std::fs::write(&path, serde_yaml::to_string(&config).unwrap()).unwrap(); child.signal(libc::SIGHUP); child.wait_for("Servers ready").await; diff --git a/tests/cases/process_contract/clash_api.rs b/tests/cases/process_contract/clash_api.rs index 22c43d03..be76bf82 100644 --- a/tests/cases/process_contract/clash_api.rs +++ b/tests/cases/process_contract/clash_api.rs @@ -383,6 +383,51 @@ async fn url_test_routes_without_controller_and_can_be_removed_on_reload() { std::net::TcpListener::bind(fixture.entry).unwrap(); } +#[tokio::test] +async fn failed_activation_restores_control_generation_and_selection() { + let mut fixture = Fixture::new().await; + let mut process = fixture.start().await; + let mut old = flow(fixture.entry, b'E').await; + let generation = api(fixture.api, "GET", "/connections", Value::Null).await.1["connections"][0] + ["shoesGeneration"] + .clone(); + let live = fixture.config.clone(); + fixture.outbound_mut()["default"] = json!("west"); + let partial = available_address(); + fixture.config.as_array_mut().unwrap().extend([ + json!({"address":partial.to_string(), "protocol":{"type":"socks"}}), + json!({"path":fixture.directory.path().join("missing/listener.sock"), "protocol":{"type":"socks"}}), + ]); + fixture.write(); + process.signal(libc::SIGHUP); + process + .wait_for("Reload rejected; previous servers restored") + .await; + assert_eq!(fixture.selected().await, "east"); + check_flow(&mut old, b'E').await; + let mut restored = flow(fixture.entry, b'E').await; + let snapshot = api(fixture.api, "GET", "/connections", Value::Null).await.1; + assert!( + snapshot["connections"] + .as_array() + .unwrap() + .iter() + .all(|flow| flow["shoesGeneration"] == generation) + ); + drop(std::net::TcpListener::bind(partial).unwrap()); + fixture.select("west").await; + check_flow(&mut restored, b'E').await; + flow(fixture.entry, b'W').await; + + fixture.config = live; + fixture.write(); + process.signal(libc::SIGHUP); + process.wait_for("Servers ready").await; + assert_eq!(fixture.selected().await, "west"); + flow(fixture.entry, b'W').await; + fixture.stop(&mut process).await; +} + #[tokio::test] async fn selectors_change_real_traffic_and_restore_after_restart() { let mut fixture = Fixture::new().await; diff --git a/tests/cases/process_contract/fake_dns.rs b/tests/cases/process_contract/fake_dns.rs index 06dd2f0e..191fc8f9 100644 --- a/tests/cases/process_contract/fake_dns.rs +++ b/tests/cases/process_contract/fake_dns.rs @@ -35,7 +35,7 @@ async fn fake_dns_reload_shutdown_and_crash_preserve_safe_mapping_state() { let port = available_address().port(); let mut config = json!([ {"dns_group":"shared", "dns_servers":"udp://127.0.0.1:9", - "fake_ip":{"state_file":state,"max_entries":16}}, + "fake_ip":{"state_file":state,"max_entries":16,"ipv4_range":"198.18.0.0/24"}}, {"address":format!("0.0.0.0:{port}"),"protocol":{"type":"socks","udp_enabled":true}, "dns":{"servers":"shared"}} ]); @@ -51,6 +51,21 @@ async fn fake_dns_reload_shutdown_and_crash_preserve_safe_mapping_state() { let mut process = Process::start(&path, &["--no-reload"]); process.wait_for("Servers ready").await; let stable = query(port, "stable.example.").await; + let mut failed = config.clone(); + failed.as_array_mut().unwrap().extend([ + json!({"dns_group":"candidate", "dns_servers":"system", "fake_ip":{ + "state_file":directory.path().join("candidate.json"), + "ipv4_range":"198.19.0.0/24", "ipv6_range":"2001:2:1::/64", "max_entries":16 + }}), + json!({"path":directory.path().join("missing/listener.sock"), "protocol":{"type":"socks"}}), + ]); + std::fs::write(&path, failed.to_string()).unwrap(); + process.signal(libc::SIGHUP); + process.wait_for("Reload rejected; previous servers restored").await; + assert_eq!(query(port, "stable.example.").await, stable); + query(port, "after-rejection.example.").await; + failed.as_array_mut().unwrap().pop(); + config = failed; config[0]["fake_ip"]["https_records"] = "suppress".into(); std::fs::write(&path, config.to_string()).unwrap(); process.signal(libc::SIGHUP); diff --git a/tests/cases/retention/process.rs b/tests/cases/retention/process.rs index 97f030dd..e602b4e4 100644 --- a/tests/cases/retention/process.rs +++ b/tests/cases/retention/process.rs @@ -196,8 +196,12 @@ async fn exercise() { }) .await; let mut candidate = prepare(&paths, Some(&prepared.dns)).await.unwrap(); + candidate.tcp = + TcpListeners::prepare(&candidate.validated.configs, &prepared.tcp, &[]).unwrap(); running.stop().await; + prepared.tcp.retain_shared(&candidate.tcp); running.start(&mut candidate).await.unwrap(); + candidate.tcp.activate(); drop(prepared); echo(&mut live, cycle).await; let mut replacement = connect(forward).await; diff --git a/tests/process_contract.rs b/tests/process_contract.rs index 70aebabd..b71bfe32 100644 --- a/tests/process_contract.rs +++ b/tests/process_contract.rs @@ -1148,6 +1148,23 @@ async fn connection_budget_changes_only_with_accepted_reloads_and_counts_drainin assert_rejected(address).await; roundtrip(&mut draining).await; + // Activation applies the candidate cap, then fails after starting its TCP listener. + std::fs::write( + &path, + format!( + "{}- path: '{}'\n protocol: {{type: http}}\n", + config(2), + directory.path().join("missing/listener.sock").display(), + ), + ) + .unwrap(); + child.signal(libc::SIGHUP); + child + .wait_for("Reload rejected; previous servers restored") + .await; + assert_rejected(address).await; + roundtrip(&mut draining).await; + // The accepted generation's budget still counts the previous generation's flow. std::fs::write(&path, config(2)).unwrap(); child.signal(libc::SIGHUP); @@ -1470,24 +1487,210 @@ async fn shutdown_interrupts_dns_preparation_without_retiring_listeners() { #[cfg(unix)] #[tokio::test] -async fn replacement_bind_failure_is_fatal_and_cleans_partial_generation() { +async fn replacement_bind_failure_retains_service_and_allows_subsequent_reload() { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("config.yaml"); + let echo = shoes_test_support::test_servers::start_tcp_stream_echo_server("127.0.0.1", 0) + .await + .unwrap(); let address = available_address(); + let candidate_address = available_address(); let occupied = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let config = format!("- address: '{address}'\n protocol: {{type: http}}\n"); + let config = forward(address, echo.local_addr()); std::fs::write(&path, &config).unwrap(); let mut child = Process::start(&path, &["--no-reload"]); child.wait_for("Servers ready").await; + let mut retained = TcpStream::connect(address).await.unwrap(); + roundtrip(&mut retained).await; std::fs::write( &path, format!( - "{config}- address: '{}'\n protocol: {{type: http}}\n", + "{config}- address: '{candidate_address}'\n protocol: {{type: http}}\n- address: '{}'\n protocol: {{type: http}}\n", occupied.local_addr().unwrap() ), ) .unwrap(); child.signal(libc::SIGHUP); + child + .wait_for("Reload rejected; current servers retained") + .await; + roundtrip(&mut retained).await; + roundtrip(&mut TcpStream::connect(address).await.unwrap()).await; + drop(std::net::TcpListener::bind(candidate_address).unwrap()); + std::fs::write( + &path, + format!("{config}- address: '{candidate_address}'\n protocol: {{type: http}}\n"), + ) + .unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Servers ready").await; + roundtrip(&mut retained).await; + roundtrip(&mut TcpStream::connect(address).await.unwrap()).await; + TcpStream::connect(candidate_address).await.unwrap(); + child.signal(libc::SIGTERM); + assert_eq!(child.exit().await.code(), Some(0)); + let socket = tokio::net::TcpSocket::new_v4().unwrap(); + socket.set_reuseaddr(true).unwrap(); + socket.bind(address).unwrap(); + socket.listen(128).unwrap(); +} + +#[cfg(unix)] +#[tokio::test] +async fn partial_quic_activation_rolls_back_or_shuts_down_without_leaking_listeners() { + for cancel in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.yaml"); + let echo = shoes_test_support::test_servers::start_tcp_stream_echo_server("127.0.0.1", 0) + .await + .unwrap(); + let address = available_address(); + let live = forward(address, echo.local_addr()) + .replace("reload_grace_secs: 2", "reload_grace_secs: 60"); + std::fs::write(&path, &live).unwrap(); + let mut child = Process::start(&path, &["--no-reload"]); + child.wait_for("Servers ready").await; + let mut retained = TcpStream::connect(address).await.unwrap(); + roundtrip(&mut retained).await; + let reservation = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let partial = reservation.local_addr().unwrap(); + let occupied = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let blocked = occupied.local_addr().unwrap(); + drop(reservation); + let certificate = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let quic = |address: SocketAddr| { + serde_json::json!({ + "address": address.to_string(), "transport":"quic", + "quic_settings":{"cert":certificate.cert.pem(), "key":certificate.signing_key.serialize_pem(), "num_endpoints":1}, + "protocol":{"type":"forward", "target":echo.local_addr().to_string()} + }) + }; + let mut candidate: serde_json::Value = serde_yaml::from_str(&live).unwrap(); + candidate + .as_array_mut() + .unwrap() + .extend([quic(partial), quic(blocked)]); + std::fs::write(&path, candidate.to_string()).unwrap(); + child.signal(libc::SIGHUP); + if cancel { + child.wait_for(&format!("QUIC server at {blocked}")).await; + } else { + child + .wait_for("Reload rejected; previous servers restored") + .await; + roundtrip(&mut retained).await; + roundtrip(&mut TcpStream::connect(address).await.unwrap()).await; + drop(std::net::UdpSocket::bind(partial).unwrap()); + candidate.as_array_mut().unwrap().pop(); + std::fs::write(&path, candidate.to_string()).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Servers ready").await; + } + child.signal(libc::SIGTERM); + assert!(child.exit().await.success()); + drop(std::net::UdpSocket::bind(partial).unwrap()); + let socket = tokio::net::TcpSocket::new_v4().unwrap(); + socket.set_reuseaddr(true).unwrap(); + socket.bind(address).unwrap(); + socket.listen(128).unwrap(); + } +} + +#[cfg(unix)] +#[tokio::test] +async fn descriptor_tun_can_restart_after_failed_activation_and_releases_on_exit() { + check_descriptor_tun_reload(true).await; +} + +#[cfg(unix)] +#[tokio::test] +async fn candidate_only_tun_descriptor_survives_rejected_reload() { + check_descriptor_tun_reload(false).await; +} + +#[cfg(unix)] +async fn check_descriptor_tun_reload(initially_enabled: bool) { + use std::os::fd::OwnedFd; + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.yaml"); + let echo = shoes_test_support::test_servers::start_tcp_stream_echo_server("127.0.0.1", 0) + .await + .unwrap(); + let address = available_address(); + let config = format!( + "{}- device_fd: 0\n rules: allow-all-direct\n", + forward(address, echo.local_addr()) + ); + let initial = if initially_enabled { + config.clone() + } else { + forward(address, echo.local_addr()) + }; + std::fs::write(&path, initial).unwrap(); + let (mut peer, device) = std::os::unix::net::UnixStream::pair().unwrap(); + peer.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_shoes")) + .args(["-t", "1", "--no-reload"]) + .arg(&path) + .stdin(Stdio::from(OwnedFd::from(device))) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut child = Process::from_child(child); + child.wait_for("Servers ready").await; + // Proxy entries start first; a second TUN forces failure after descriptor adoption. + std::fs::write( + &path, + format!("{config}- device_fd: -1\n rules: allow-all-direct\n"), + ) + .unwrap(); + child.signal(libc::SIGHUP); + child + .wait_for("Reload rejected; previous servers restored") + .await; + let error = std::io::Read::read(&mut peer, &mut [0]) + .expect_err("rejected reload closed the inherited TUN descriptor"); + assert!(matches!( + error.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + )); + roundtrip(&mut TcpStream::connect(address).await.unwrap()).await; + std::fs::write(&path, config).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Servers ready").await; + std::fs::write(&path, forward(address, echo.local_addr())).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("Servers ready").await; + assert_eq!(std::io::Read::read(&mut peer, &mut [0]).unwrap(), 0); + child.signal(libc::SIGTERM); + assert!(child.exit().await.success()); +} + +#[cfg(unix)] +#[tokio::test] +async fn rollback_failure_is_reported_and_partial_services_are_stopped() { + let directory = tempfile::tempdir().unwrap(); + let parent = directory.path().join("old"); + std::fs::create_dir(&parent).unwrap(); + let socket = parent.join("listener.sock"); + let path = directory.path().join("config.yaml"); + std::fs::write( + &path, + format!( + "- path: '{}'\n protocol: {{type: socks}}\n", + socket.display() + ), + ) + .unwrap(); + let mut child = Process::start(&path, &["--no-reload"]); + child.wait_for("Servers ready").await; + std::fs::remove_file(&socket).unwrap(); + std::fs::remove_dir(parent).unwrap(); + let partial = available_address(); + std::fs::write(&path, format!("- address: '{partial}'\n protocol: {{type: socks}}\n- path: '{}'\n protocol: {{type: socks}}\n", directory.path().join("missing/listener.sock").display())).unwrap(); + child.signal(libc::SIGHUP); + child.wait_for("rollback failed").await; assert_eq!(child.exit().await.code(), Some(1)); - std::net::TcpListener::bind(address).unwrap(); + std::net::TcpListener::bind(partial).unwrap(); }