From cd02cb0e6dacecd10df1f4f0cd37ffda627e9bf3 Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Thu, 2 Jul 2026 18:16:11 +0400 Subject: [PATCH 1/6] fix(proxy): graceful drain on Windows via named kernel event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rotation on Windows was a hard TerminateProcess: graceful_stop_child had no non-unix drain path, and the worker only listened for Ctrl+C. Every network-change rotation reset all in-flight connections — the top cause of "hotspot/AP works badly on Windows". Worker (soth-proxy): new drain_signal module. On startup the worker creates a named kernel event Local\soth-worker-drain-{pid} and waits on it from a detached thread (GenerateConsoleCtrlEvent cannot cross the CREATE_NO_WINDOW console boundary; a named event can). Event fires -> same graceful path SIGUSR1 takes on Unix (stop accepting, drain 30s). Both duplicated shutdown-wait blocks in runtime.rs now share drain_signal::wait_for_shutdown_signal(). Supervisor (soth-cli): Windows graceful_stop_child opens the child's drain event, sets it, and waits up to 30s for exit — mirroring the Unix SIGUSR1 flow. Falls back to the old hard kill when the event can't be opened (older worker, dead child) or the drain window elapses. Part of docs/common/2026-07-02/network-resilience-impl-plan.md (Phase 2). Co-Authored-By: Claude Fable 5 --- Cargo.lock | 1 + crates/soth-cli/Cargo.toml | 1 + crates/soth-cli/src/commands/proxy/start.rs | 52 +++++++++- crates/soth-proxy/Cargo.toml | 7 ++ crates/soth-proxy/src/drain_signal.rs | 108 ++++++++++++++++++++ crates/soth-proxy/src/lib.rs | 1 + crates/soth-proxy/src/runtime.rs | 41 +------- 7 files changed, 173 insertions(+), 38 deletions(-) create mode 100644 crates/soth-proxy/src/drain_signal.rs diff --git a/Cargo.lock b/Cargo.lock index 71e510d..b956442 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4406,6 +4406,7 @@ dependencies = [ "tracing-opentelemetry", "tracing-subscriber", "uuid", + "windows-sys 0.59.0", "zeroize", ] diff --git a/crates/soth-cli/Cargo.toml b/crates/soth-cli/Cargo.toml index 6c7a32d..70a3792 100644 --- a/crates/soth-cli/Cargo.toml +++ b/crates/soth-cli/Cargo.toml @@ -59,6 +59,7 @@ comfy-table = "7.0" windows-sys = { version = "0.59", features = [ "Win32_Foundation", "Win32_Networking_WinInet", + "Win32_System_Threading", ] } [dev-dependencies] diff --git a/crates/soth-cli/src/commands/proxy/start.rs b/crates/soth-cli/src/commands/proxy/start.rs index c2dbf9a..f9378bd 100644 --- a/crates/soth-cli/src/commands/proxy/start.rs +++ b/crates/soth-cli/src/commands/proxy/start.rs @@ -981,7 +981,57 @@ async fn graceful_stop_child(child: &mut Child) -> Result<()> { Ok(()) } -#[cfg(not(unix))] +/// Windows equivalent of the SIGUSR1 drain: sets the named kernel event the +/// worker created at startup (`Local\soth-worker-drain-{pid}`, see +/// `soth_proxy::drain_signal`), then waits up to 30 seconds for exit. +/// Falls back to a hard kill if the event can't be opened (worker predates +/// this protocol, or died) or the drain window elapses. +#[cfg(windows)] +async fn graceful_stop_child(child: &mut Child) -> Result<()> { + let signaled = child.id().map(signal_windows_drain_event).unwrap_or(false); + if !signaled { + warn!("could not signal drain event on old proxy child; falling back to hard kill"); + return terminate_child(child).await; + } + match tokio::time::timeout(Duration::from_secs(30), child.wait()).await { + Ok(Ok(status)) => { + info!(status = %status, "old proxy child exited after drain"); + } + Ok(Err(error)) => { + warn!(error = %error, "error waiting for old proxy child"); + } + Err(_) => { + warn!("old proxy child did not exit within 30s drain window; killing"); + let _ = terminate_child(child).await; + } + } + Ok(()) +} + +/// Open the worker's named drain event and set it. Returns `false` when the +/// event doesn't exist or can't be signaled — callers fall back to a hard +/// kill, matching the pre-drain behavior. +#[cfg(windows)] +fn signal_windows_drain_event(pid: u32) -> bool { + use windows_sys::Win32::Foundation::CloseHandle; + use windows_sys::Win32::System::Threading::{OpenEventW, SetEvent, EVENT_MODIFY_STATE}; + + let name: Vec = soth_proxy::drain_signal::drain_event_name(pid) + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + unsafe { + let handle = OpenEventW(EVENT_MODIFY_STATE, 0, name.as_ptr()); + if handle.is_null() { + return false; + } + let signaled = SetEvent(handle) != 0; + CloseHandle(handle); + signaled + } +} + +#[cfg(not(any(unix, windows)))] async fn graceful_stop_child(child: &mut Child) -> Result<()> { terminate_child(child).await } diff --git a/crates/soth-proxy/Cargo.toml b/crates/soth-proxy/Cargo.toml index ca46e0d..9e119ad 100644 --- a/crates/soth-proxy/Cargo.toml +++ b/crates/soth-proxy/Cargo.toml @@ -64,6 +64,13 @@ soth-sync = { workspace = true } soth-extensions = { workspace = true, optional = true } soth-mitm = "0.3.3" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_System_Threading", +] } + [dev-dependencies] tower = { version = "0.5", features = ["util"] } tempfile = { workspace = true } diff --git a/crates/soth-proxy/src/drain_signal.rs b/crates/soth-proxy/src/drain_signal.rs new file mode 100644 index 0000000..189b727 --- /dev/null +++ b/crates/soth-proxy/src/drain_signal.rs @@ -0,0 +1,108 @@ +//! Cross-platform graceful-drain signal for the proxy worker. +//! +//! Unix: SIGUSR1, sent by the supervisor during graceful child rotation. +//! Windows: a named kernel event `Local\soth-worker-drain-{pid}` that the +//! supervisor opens and sets. `GenerateConsoleCtrlEvent` is not usable here — +//! the worker is spawned with `CREATE_NO_WINDOW` so it does not share a +//! console with the supervisor, and console ctrl events cannot cross that +//! boundary. A named event has no such restriction. +//! +//! Without this, Windows rotation had no drain path at all: the supervisor +//! hard-killed the worker (`TerminateProcess`), resetting every in-flight +//! connection on each network change. + +use tracing::info; + +/// The name of the drain event for a worker process id. Kept in sync with +/// the supervisor side (`soth-cli/src/commands/proxy/start.rs`). +#[cfg(windows)] +pub fn drain_event_name(pid: u32) -> String { + format!("Local\\soth-worker-drain-{pid}") +} + +/// Wait until a shutdown is requested: Ctrl+C on all platforms, plus +/// SIGUSR1 (Unix) or the named drain event (Windows). +pub(crate) async fn wait_for_shutdown_signal() { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); + tokio::select! { + _ = tokio::signal::ctrl_c() => { + info!("shutdown requested (Ctrl+C)"); + } + _ = usr1.recv() => { + info!("graceful drain requested (SIGUSR1)"); + } + } + } + #[cfg(windows)] + { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + // A plain std thread (not spawn_blocking): the tokio runtime waits + // for blocking tasks on drop, and this thread parks forever when + // shutdown comes via Ctrl+C instead of the event. + let spawned = std::thread::Builder::new() + .name("soth-drain-event".to_string()) + .spawn(move || { + if wait_for_drain_event(std::process::id()) { + let _ = tx.send(()); + } + // On failure the sender drops; the receiver arm below + // falls back to Ctrl+C-only. + }) + .is_ok(); + if !spawned { + tracing::warn!("failed to spawn drain-event thread; Ctrl+C only"); + } + tokio::select! { + _ = tokio::signal::ctrl_c() => { + info!("shutdown requested (Ctrl+C)"); + } + result = rx => match result { + Ok(()) => { + info!("graceful drain requested (drain event)"); + } + Err(_) => { + tracing::warn!( + "drain-event listener unavailable; waiting on Ctrl+C only" + ); + let _ = tokio::signal::ctrl_c().await; + info!("shutdown requested (Ctrl+C)"); + } + }, + } + } + #[cfg(not(any(unix, windows)))] + { + let _ = tokio::signal::ctrl_c().await; + info!("shutdown requested (Ctrl+C)"); + } +} + +/// Create the named drain event and block until the supervisor sets it. +/// Returns `true` when the event fired, `false` on any setup/wait failure. +#[cfg(windows)] +fn wait_for_drain_event(pid: u32) -> bool { + use windows_sys::Win32::Foundation::{CloseHandle, WAIT_OBJECT_0}; + use windows_sys::Win32::System::Threading::{CreateEventW, WaitForSingleObject, INFINITE}; + + let name: Vec = drain_event_name(pid) + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + // Manual-reset, initially unsignaled. Created by the worker (not the + // supervisor) so it exists for the worker's whole lifetime; the + // supervisor opens it by name only when it wants to drain. + let handle = unsafe { CreateEventW(std::ptr::null(), 1, 0, name.as_ptr()) }; + if handle.is_null() { + tracing::warn!( + error = %std::io::Error::last_os_error(), + "failed to create drain event; graceful drain unavailable" + ); + return false; + } + let wait_result = unsafe { WaitForSingleObject(handle, INFINITE) }; + unsafe { CloseHandle(handle) }; + wait_result == WAIT_OBJECT_0 +} diff --git a/crates/soth-proxy/src/lib.rs b/crates/soth-proxy/src/lib.rs index a19a8ca..dffc9ba 100644 --- a/crates/soth-proxy/src/lib.rs +++ b/crates/soth-proxy/src/lib.rs @@ -14,6 +14,7 @@ pub mod classify_task; pub mod config; pub mod db; +pub mod drain_signal; pub mod error; pub mod gating; pub mod runtime; diff --git a/crates/soth-proxy/src/runtime.rs b/crates/soth-proxy/src/runtime.rs index cc830d2..be26372 100644 --- a/crates/soth-proxy/src/runtime.rs +++ b/crates/soth-proxy/src/runtime.rs @@ -2,7 +2,8 @@ //! //! `run(registry)` drives the full proxy lifecycle: loads config, opens the DB, //! downloads classify models if needed, spins up MITM + ops + sync + telemetry, -//! then waits for Ctrl+C or SIGUSR1 (graceful drain) and shuts everything down. +//! then waits for Ctrl+C or a graceful-drain signal (SIGUSR1 on Unix, the +//! named drain event on Windows — see `drain_signal`) and shuts everything down. //! //! Extension registration is the caller's responsibility: the `soth` CLI //! constructs an `ExtensionRegistry` (registering e.g. `HistorianExtension`) @@ -338,24 +339,7 @@ async fn run_inner(ext_registry: Option) -> Result<()> { info!("soth-proxy started; press Ctrl+C to stop"); - #[cfg(unix)] - { - use tokio::signal::unix::{signal, SignalKind}; - let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); - tokio::select! { - _ = tokio::signal::ctrl_c() => { - info!("shutdown requested (Ctrl+C)"); - } - _ = usr1.recv() => { - info!("graceful drain requested (SIGUSR1)"); - } - } - } - #[cfg(not(unix))] - { - tokio::signal::ctrl_c().await.context("wait for Ctrl+C")?; - info!("shutdown requested"); - } + crate::drain_signal::wait_for_shutdown_signal().await; proxy_handle .shutdown(Duration::from_secs(30)) .await @@ -649,24 +633,7 @@ async fn run_inner() -> Result<()> { info!("soth-proxy started; press Ctrl+C to stop"); - #[cfg(unix)] - { - use tokio::signal::unix::{signal, SignalKind}; - let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); - tokio::select! { - _ = tokio::signal::ctrl_c() => { - info!("shutdown requested (Ctrl+C)"); - } - _ = usr1.recv() => { - info!("graceful drain requested (SIGUSR1)"); - } - } - } - #[cfg(not(unix))] - { - tokio::signal::ctrl_c().await.context("wait for Ctrl+C")?; - info!("shutdown requested"); - } + crate::drain_signal::wait_for_shutdown_signal().await; proxy_handle .shutdown(Duration::from_secs(30)) .await From 4e22ce28e5904b70efef52a5b9739c6422c68bf5 Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Thu, 2 Jul 2026 18:23:20 +0400 Subject: [PATCH 2/6] =?UTF-8?q?fix(cli):=20network=20watcher=20v2=20?= =?UTF-8?q?=E2=80=94=20IPv6=20+=20DNS=20signature,=20offline-recovery=20ed?= =?UTF-8?q?ge,=20debounce?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The watcher compared only the IPv4 address set (Unix) or the default-route IPv4 (Windows UDP-connect trick). Hotspot/AP transitions it missed entirely: - toggle hotspot off/on or roam APs: same IP comes back, no event — the worker kept its startup-pinned DNS nameservers and dead sockets ("tunnel connection failed" until `soth up`) - IPv6-only / IPv6-primary tethering: invisible - DNS-server change without an address change: invisible - offline ticks were skipped outright, so drop -> recover-same-address never fired Replace with a NetworkSignature {v4, v6 (global+ULA), dns} poll: - Unix: getifaddrs extended to AF_INET6; DNS parsed from /etc/resolv.conf (configd-managed on macOS; inert-but-harmless on systemd-resolved stubs) - Windows: adapter enumeration via the ipconfig crate (what hickory-resolver uses), replacing the UDP trick — covers addresses, IPv6, and per-adapter DNS in one call - offline marker: recovery fires even with an identical signature - 10s debounce with trailing delivery: a flapping hotspot causes one rotation per window instead of a rotation storm Watcher core now takes an injectable signature source; new tests cover offline-recovery, DNS-only change, and debounce/trailing behavior. 117 soth-cli unit tests pass. Part of docs/common/2026-07-02/network-resilience-impl-plan.md (Phase 3). Co-Authored-By: Claude Fable 5 --- Cargo.lock | 1 + crates/soth-cli/Cargo.toml | 1 + .../src/commands/proxy/network_watcher.rs | 440 ++++++++++++++---- 3 files changed, 359 insertions(+), 83 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b956442..0e5fd36 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4093,6 +4093,7 @@ dependencies = [ "dirs", "ed25519-dalek", "hex", + "ipconfig", "libc", "owo-colors", "rand 0.8.5", diff --git a/crates/soth-cli/Cargo.toml b/crates/soth-cli/Cargo.toml index 70a3792..2523711 100644 --- a/crates/soth-cli/Cargo.toml +++ b/crates/soth-cli/Cargo.toml @@ -61,6 +61,7 @@ windows-sys = { version = "0.59", features = [ "Win32_Networking_WinInet", "Win32_System_Threading", ] } +ipconfig = "0.3" [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } diff --git a/crates/soth-cli/src/commands/proxy/network_watcher.rs b/crates/soth-cli/src/commands/proxy/network_watcher.rs index 33d12a9..d5959c8 100644 --- a/crates/soth-cli/src/commands/proxy/network_watcher.rs +++ b/crates/soth-cli/src/commands/proxy/network_watcher.rs @@ -1,46 +1,108 @@ //! Detects primary-network changes (wifi switch, captive-portal address -//! reassignment, VPN up/down) and signals the supervisor to graceful-rotate -//! the proxy worker. Without this, after a network change `soth-mitm`'s -//! upstream connection pool keeps trying to reuse TCP sockets bound to the -//! old gateway — surfacing as "tunnel connection failed" until `soth up`. +//! reassignment, VPN up/down, hotspot toggles) and signals the supervisor to +//! graceful-rotate the proxy worker. Without this, after a network change +//! `soth-mitm`'s DNS resolver and upstream sockets keep pointing at the old +//! gateway — surfacing as "tunnel connection failed" until `soth up`. //! -//! Implementation: poll `getifaddrs(3)` every 5s and compare the set of -//! non-loopback IPv4 addresses bound to UP interfaces. Polling (vs. -//! event-driven `SCNetworkConfiguration` / netlink / `NotifyIpInterfaceChange`) -//! costs us a few seconds of latency but avoids three platform-specific code -//! paths and a new dependency. The symptom we're fixing already takes -//! seconds to surface, so the latency floor is acceptable. +//! Implementation: poll a [`NetworkSignature`] every 5s and compare. The +//! signature covers what actually changes across hotspot/AP transitions: +//! +//! - non-loopback IPv4 **and** IPv6 addresses on UP interfaces (v6-only and +//! v6-primary tethering used to be invisible), +//! - the system DNS server list (a same-subnet network switch can change +//! resolvers without changing our addresses), +//! - an offline marker, so *drop → recover on the same address* (hotspot +//! toggled off/on, AP roam) fires a rotation even though the signature is +//! unchanged — previously the empty set was skipped and the recovery was +//! silent. +//! +//! Fires are debounced: at most one signal per [`DEBOUNCE`] window, with the +//! trailing change delivered on the first tick after the window closes, so a +//! flapping hotspot causes one rotation per window instead of a storm. +//! +//! Polling (vs. event-driven `SCNetworkConfiguration` / netlink / +//! `NotifyIpInterfaceChange`) costs a few seconds of latency but avoids +//! three platform-specific code paths. The symptom we're fixing already +//! takes seconds to surface, so the latency floor is acceptable. use std::collections::BTreeSet; -use std::net::Ipv4Addr; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::time::Duration; use tokio::sync::watch; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; const POLL_INTERVAL: Duration = Duration::from_secs(5); +/// Minimum spacing between emitted change signals. Rotation itself takes a +/// few seconds; anything faster just kills connections repeatedly. +const DEBOUNCE: Duration = Duration::from_secs(10); + +/// Point-in-time view of the network facts the proxy worker depends on. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +struct NetworkSignature { + v4: BTreeSet, + /// Global + ULA only. Link-local (fe80::/10) churns on every interface + /// event without affecting routing and is excluded. + v6: BTreeSet, + /// Ordered system DNS server list. + dns: Vec, +} + +impl NetworkSignature { + /// No usable addresses — treat as offline regardless of DNS remnants. + fn is_addrless(&self) -> bool { + self.v4.is_empty() && self.v6.is_empty() + } +} /// Spawn the watcher. Returns a `watch::Receiver` whose value increments on /// every detected change. The supervisor awaits `changed()` on it. pub fn spawn() -> watch::Receiver { + spawn_with_source(read_network_signature) +} + +/// Watcher core with an injectable signature source (tests script network +/// transitions through it; production passes [`read_network_signature`]). +fn spawn_with_source( + mut source: impl FnMut() -> NetworkSignature + Send + 'static, +) -> watch::Receiver { let (tx, rx) = watch::channel(0u64); tokio::spawn(async move { - let mut last = local_v4_addrs(); - debug!(addrs = ?last, "network watcher: initial address set"); + let mut last = source(); + debug!(signature = ?last, "network watcher: initial signature"); + let mut was_offline = false; + let mut pending_signal = false; + let mut last_fire: Option = None; let mut tick = tokio::time::interval(POLL_INTERVAL); tick.tick().await; loop { tick.tick().await; - let current = local_v4_addrs(); - if current.is_empty() { + let current = source(); + if current.is_addrless() { + // Offline. Remember it so recovery fires even when the + // network comes back with an identical signature. + if !was_offline { + debug!("network watcher: all addresses gone (offline)"); + } + was_offline = true; continue; } - if current != last { + if current != last || was_offline { info!( previous = ?last, current = ?current, - "primary network changed — signalling supervisor for graceful proxy rotation" + recovered_from_offline = was_offline, + "primary network changed — scheduling supervisor signal for graceful proxy rotation" ); last = current; + was_offline = false; + pending_signal = true; + } + let debounce_open = last_fire + .map(|fired| fired.elapsed() >= DEBOUNCE) + .unwrap_or(true); + if pending_signal && debounce_open { + pending_signal = false; + last_fire = Some(tokio::time::Instant::now()); let next = tx.borrow().wrapping_add(1); if tx.send(next).is_err() { return; @@ -51,15 +113,38 @@ pub fn spawn() -> watch::Receiver { rx } +fn read_network_signature() -> NetworkSignature { + let (v4, v6) = local_addrs(); + NetworkSignature { + v4, + v6, + dns: system_dns_servers(), + } +} + +/// True for IPv6 addresses that identify a network rather than interface +/// noise: global unicast and ULA, excluding loopback/link-local/unspecified. +fn is_meaningful_v6(addr: &Ipv6Addr) -> bool { + if addr.is_loopback() || addr.is_unspecified() { + return false; + } + // Link-local fe80::/10. + (addr.segments()[0] & 0xffc0) != 0xfe80 +} + #[cfg(unix)] -fn local_v4_addrs() -> BTreeSet { - use libc::{freeifaddrs, getifaddrs, ifaddrs, sockaddr_in, AF_INET, IFF_LOOPBACK, IFF_UP}; - let mut set = BTreeSet::new(); +fn local_addrs() -> (BTreeSet, BTreeSet) { + use libc::{ + freeifaddrs, getifaddrs, ifaddrs, sockaddr_in, sockaddr_in6, AF_INET, AF_INET6, + IFF_LOOPBACK, IFF_UP, + }; + let mut v4 = BTreeSet::new(); + let mut v6 = BTreeSet::new(); unsafe { let mut head: *mut ifaddrs = std::ptr::null_mut(); if getifaddrs(&mut head) != 0 || head.is_null() { - warn!("getifaddrs failed; skipping network change check this tick"); - return set; + tracing::warn!("getifaddrs failed; skipping network change check this tick"); + return (v4, v6); } let mut cur = head; while !cur.is_null() { @@ -72,71 +157,118 @@ fn local_v4_addrs() -> BTreeSet { if flags & IFF_LOOPBACK != 0 || flags & IFF_UP == 0 { continue; } - if (*entry.ifa_addr).sa_family as i32 != AF_INET { - continue; + match (*entry.ifa_addr).sa_family as i32 { + family if family == AF_INET => { + let sin = &*(entry.ifa_addr as *const sockaddr_in); + v4.insert(Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr))); + } + family if family == AF_INET6 => { + let sin6 = &*(entry.ifa_addr as *const sockaddr_in6); + let addr = Ipv6Addr::from(sin6.sin6_addr.s6_addr); + if is_meaningful_v6(&addr) { + v6.insert(addr); + } + } + _ => {} } - let sin = &*(entry.ifa_addr as *const sockaddr_in); - let addr = Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr)); - set.insert(addr); } freeifaddrs(head); } - set + (v4, v6) } -/// Windows path: ask the kernel which local IPv4 it would route to a -/// public destination via a UDP "connect" trick. Connecting a UDP -/// socket sends no packets — it only updates the kernel's route -/// resolution for that socket — but `local_addr()` then returns the -/// IP that the default route would use. That IP is exactly what -/// flips on a wifi switch / interface change, so polling it -/// detects the change without any FFI into `GetAdaptersAddresses`. -/// -/// Trade-off vs `getifaddrs` on Unix: this captures only the -/// default-route IP, not the full set of bound IPv4s. On Windows -/// that's enough for the watcher's purpose — the supervisor reload -/// is triggered by *any* change, and the default-route IP changes -/// on every realistic network event (wifi network switch, VPN -/// up/down, dock/undock with USB-Ethernet, captive-portal IP -/// reassignment). It does NOT catch "added a secondary adapter -/// while the existing default route is still active" but neither -/// did the previous empty stub, and the upstream connection pool -/// only cares about routes that actually flip. +/// Windows: enumerate adapters via `ipconfig` (the same crate +/// hickory-resolver uses for its Windows DNS discovery). This replaces the +/// old UDP-connect trick, which captured only the default-route IPv4 — it +/// missed secondary adapters, every IPv6 transition, and DNS-only changes. #[cfg(windows)] -fn local_v4_addrs() -> BTreeSet { - use std::net::{IpAddr, UdpSocket}; +fn local_addrs() -> (BTreeSet, BTreeSet) { + let mut v4 = BTreeSet::new(); + let mut v6 = BTreeSet::new(); + let Ok(adapters) = ipconfig::get_adapters() else { + tracing::warn!("ipconfig::get_adapters failed; skipping network change check this tick"); + return (v4, v6); + }; + for adapter in adapters + .iter() + .filter(|adapter| adapter.oper_status() == ipconfig::OperStatus::IfOperStatusUp) + { + for addr in adapter.ip_addresses() { + match addr { + IpAddr::V4(value) => { + if !value.is_loopback() && !value.is_unspecified() { + v4.insert(*value); + } + } + IpAddr::V6(value) => { + if is_meaningful_v6(value) { + v6.insert(*value); + } + } + } + } + } + (v4, v6) +} + +/// Catch-all for non-Unix, non-Windows targets (BSDs we don't officially +/// ship to, WASM, etc.). Empty sets mean the watcher polls but never fires. +#[cfg(not(any(unix, windows)))] +fn local_addrs() -> (BTreeSet, BTreeSet) { + (BTreeSet::new(), BTreeSet::new()) +} - let mut set = BTreeSet::new(); - let Ok(sock) = UdpSocket::bind("0.0.0.0:0") else { - return set; +/// System DNS servers, best-effort. An empty list disables DNS-change +/// detection for the tick but never blocks address-based detection. +/// +/// Unix: parse `/etc/resolv.conf` — written by `configd` on macOS and by +/// the resolver stack on Linux. (On systemd-resolved hosts this is the +/// constant 127.0.0.53 stub; DNS-change detection is then inert there, +/// which is harmless — address changes still fire.) Zone-scoped +/// nameservers (`fe80::1%en0`) don't parse as bare `IpAddr` and are +/// skipped. +#[cfg(unix)] +fn system_dns_servers() -> Vec { + let Ok(contents) = std::fs::read_to_string("/etc/resolv.conf") else { + return Vec::new(); }; - // 8.8.8.8 is a stable, well-known target. UDP connect doesn't - // emit packets — it only sets the destination so the kernel - // resolves a route. If the host is fully offline the connect - // can fail; fall through to an empty set, which matches the - // "no networks" baseline and won't spuriously trigger reloads. - if sock.connect("8.8.8.8:80").is_err() { - return set; - } - let Ok(addr) = sock.local_addr() else { - return set; + parse_resolv_conf_nameservers(&contents) +} + +#[cfg(unix)] +fn parse_resolv_conf_nameservers(contents: &str) -> Vec { + contents + .lines() + .filter_map(|line| { + let trimmed = line.trim(); + let value = trimmed.strip_prefix("nameserver")?.trim(); + value.split_whitespace().next()?.parse::().ok() + }) + .collect() +} + +#[cfg(windows)] +fn system_dns_servers() -> Vec { + let Ok(adapters) = ipconfig::get_adapters() else { + return Vec::new(); }; - if let IpAddr::V4(v4) = addr.ip() { - if !v4.is_loopback() && !v4.is_unspecified() { - set.insert(v4); + let mut servers = Vec::new(); + for adapter in adapters + .iter() + .filter(|adapter| adapter.oper_status() == ipconfig::OperStatus::IfOperStatusUp) + { + for server in adapter.dns_servers() { + if !servers.contains(server) { + servers.push(*server); + } } } - set + servers } -/// Catch-all for non-Unix, non-Windows targets (BSDs we don't -/// officially ship to, WASM, etc.). Empty set means the watcher -/// polls but never fires — same effective behaviour as the -/// pre-fix Windows path, just scoped to platforms we don't -/// actively support. #[cfg(not(any(unix, windows)))] -fn local_v4_addrs() -> BTreeSet { - BTreeSet::new() +fn system_dns_servers() -> Vec { + Vec::new() } #[cfg(all(test, any(unix, windows)))] @@ -144,26 +276,168 @@ mod tests { use super::*; #[test] - fn local_v4_addrs_includes_loopback_when_loopback_filter_disabled() { - // Sanity check: getifaddrs returns *something* on a normal host, and - // our filter excludes loopback. We can't assert specific addresses - // (CI / dev hosts vary), but the iteration should not panic and - // should not return loopback (127.0.0.1) since it's excluded. - let set = local_v4_addrs(); + fn local_addrs_exclude_loopback() { + let (v4, v6) = local_addrs(); + assert!( + !v4.contains(&Ipv4Addr::LOCALHOST), + "v4 loopback should be filtered out" + ); assert!( - !set.contains(&Ipv4Addr::LOCALHOST), - "loopback should be filtered out" + !v6.contains(&Ipv6Addr::LOCALHOST), + "v6 loopback should be filtered out" ); } + #[test] + fn meaningful_v6_excludes_link_local_and_loopback() { + assert!(!is_meaningful_v6(&Ipv6Addr::LOCALHOST)); + assert!(!is_meaningful_v6(&Ipv6Addr::UNSPECIFIED)); + assert!(!is_meaningful_v6(&"fe80::1".parse().unwrap())); + assert!(is_meaningful_v6(&"2001:db8::1".parse().unwrap())); + // ULA stays in — it identifies the local network. + assert!(is_meaningful_v6(&"fd00::1".parse().unwrap())); + } + + #[cfg(unix)] + #[test] + fn resolv_conf_parser_extracts_nameservers() { + let sample = "# comment\nnameserver 192.168.1.1\nnameserver 2606:4700::1111\nsearch example.com\nnameserver fe80::1%en0\nnameserver not-an-ip\n"; + let servers = parse_resolv_conf_nameservers(sample); + assert_eq!( + servers, + vec![ + "192.168.1.1".parse::().unwrap(), + "2606:4700::1111".parse::().unwrap(), + ] + ); + } + + #[test] + fn addrless_signature_reads_as_offline() { + let empty = NetworkSignature::default(); + assert!(empty.is_addrless()); + let with_dns_only = NetworkSignature { + dns: vec!["1.1.1.1".parse().unwrap()], + ..NetworkSignature::default() + }; + assert!(with_dns_only.is_addrless()); + let online = NetworkSignature { + v4: [Ipv4Addr::new(192, 168, 1, 2)].into_iter().collect(), + ..NetworkSignature::default() + }; + assert!(!online.is_addrless()); + } + #[tokio::test(flavor = "current_thread", start_paused = true)] async fn watcher_emits_no_events_when_addresses_stable() { let mut rx = spawn(); // Advance past several poll intervals; nothing should fire because - // the address set hasn't changed. + // the signature hasn't changed. tokio::time::advance(POLL_INTERVAL * 3).await; - // changed() should not have fired. let changed = tokio::time::timeout(Duration::from_millis(10), rx.changed()).await; assert!(changed.is_err(), "watcher should not emit when stable"); } + + fn sig_with_v4(last_octet: u8) -> NetworkSignature { + NetworkSignature { + v4: [Ipv4Addr::new(192, 168, 1, last_octet)] + .into_iter() + .collect(), + ..NetworkSignature::default() + } + } + + /// Drive a scripted sequence of signatures through the watcher; the + /// final entry repeats forever once the script is exhausted. + fn spawn_scripted(script: Vec) -> watch::Receiver { + let mut queue = script.into_iter(); + let mut current = NetworkSignature::default(); + spawn_with_source(move || { + if let Some(next) = queue.next() { + current = next; + } + current.clone() + }) + } + + /// Let the spawned watcher task run (paused-clock tests advance time + /// manually; the task still needs scheduler turns to consume ticks). + async fn drain_scheduler() { + for _ in 0..8 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_fires_on_offline_recovery_with_same_signature() { + // initial read, then: online → offline → online with the SAME + // signature. The old watcher skipped empty sets entirely and never + // fired on same-address recovery (hotspot toggled off/on). + let rx = spawn_scripted(vec![ + sig_with_v4(2), // initial read + sig_with_v4(2), // tick 1: stable + NetworkSignature::default(), // tick 2: offline + sig_with_v4(2), // tick 3: recovered, same address + ]); + drain_scheduler().await; + tokio::time::advance(POLL_INTERVAL * 3).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "recovery with unchanged signature must fire a rotation signal" + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_fires_on_dns_only_change() { + let base = sig_with_v4(2); + let mut with_new_dns = base.clone(); + with_new_dns.dns = vec!["9.9.9.9".parse().unwrap()]; + let rx = spawn_scripted(vec![base.clone(), base, with_new_dns]); + drain_scheduler().await; + tokio::time::advance(POLL_INTERVAL * 2).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "DNS-only change must fire" + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_debounces_rapid_flapping() { + // Flap on every tick: the first change fires immediately, changes + // inside the debounce window are deferred, and the trailing change + // is delivered once the window closes. + let mut rx = spawn_scripted(vec![ + sig_with_v4(2), // initial read + sig_with_v4(3), // tick 1: change → fires + sig_with_v4(2), // tick 2: change inside window → deferred + sig_with_v4(3), // tick 3: change inside window → still deferred + ]); + drain_scheduler().await; + + tokio::time::advance(POLL_INTERVAL).await; // tick 1 + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "first change must fire immediately" + ); + rx.borrow_and_update(); + + tokio::time::advance(POLL_INTERVAL).await; // tick 2, inside window + drain_scheduler().await; + assert!( + !rx.has_changed().expect("watcher alive"), + "change inside the debounce window must be deferred" + ); + + // Advance past the debounce window; the deferred change fires on the + // next tick. + tokio::time::advance(DEBOUNCE + POLL_INTERVAL).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "deferred change must fire once the debounce window closes" + ); + } } From e1aee8764753e7e03b40f2cc0ebf8dbb33f33812 Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Thu, 2 Jul 2026 18:46:25 +0400 Subject: [PATCH 3/6] fix(cli): unambiguous worker readiness handshake + sequenced Windows rotation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Readiness (both platforms): wait_for_listener_start TCP-probed 127.0.0.1:, which cannot distinguish the new child's listener from the old child's during rotation — and on Unix proved nothing about the worker at all, since the supervisor holds the inherited listener and the kernel completes connects from its backlog. The supervisor now passes SOTH_WORKER_READY_FILE to each spawned worker; the worker writes {"pid": ...} there (atomic temp+rename) once its listener is live, and the supervisor waits on that file. The TCP probe survives only as a late fallback (after half the startup budget) for version-skewed workers across hot-updates. Probe files are cleaned up on drop. Rotation order (Windows): with SO_EXCLUSIVEADDRUSE (soth-mitm 0.3.4) the new child can no longer bind over the old child's live listener, so the overlapped Unix flow cannot work there. Windows rotation is now sequenced: signal the drain event first (the old worker closes its listener within milliseconds and keeps draining in-flight flows), then spawn the new child into the freed port — its bind-retry loop absorbs the close latency — then wait up to 30s for the old child to finish draining. If the new child fails to start, the supervisor loop observes the draining old child's exit and respawns via the normal restart path. Part of docs/common/2026-07-02/network-resilience-impl-plan.md (Phase 4). Co-Authored-By: Claude Fable 5 --- crates/soth-cli/src/commands/proxy/start.rs | 152 ++++++++++++++++++-- crates/soth-proxy/src/drain_signal.rs | 68 +++++++++ crates/soth-proxy/src/lib.rs | 7 +- crates/soth-proxy/src/runtime.rs | 2 + 4 files changed, 211 insertions(+), 18 deletions(-) diff --git a/crates/soth-cli/src/commands/proxy/start.rs b/crates/soth-cli/src/commands/proxy/start.rs index f9378bd..df63c58 100644 --- a/crates/soth-cli/src/commands/proxy/start.rs +++ b/crates/soth-cli/src/commands/proxy/start.rs @@ -170,10 +170,11 @@ pub async fn run( #[cfg(not(unix))] let listener_fd: Option = None; - let mut child = spawn_proxy_process(generated_path.as_path(), listener_fd) + let (mut child, ready_probe) = spawn_proxy_process(generated_path.as_path(), listener_fd) .await .context("spawn soth-proxy process")?; - wait_for_listener_start(&mut child, expected_port).await?; + wait_for_listener_start(&mut child, expected_port, &ready_probe).await?; + drop(ready_probe); // Historian sibling process. Spawned only when both `enabled` and // run_mode == Subprocess. Watched in its own background task so its @@ -613,9 +614,42 @@ async fn spawn_classify_daemon_process(config_path: &Path) -> Result { .map_err(|error| anyhow::anyhow!("failed launching classify daemon worker: {error}")) } -async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Result { +/// Supervisor-side handle to the worker readiness handshake: the path the +/// spawned child will write its ready-file to (see +/// `soth_proxy::drain_signal::WORKER_READY_FILE_ENV`). +struct WorkerReadyProbe { + path: PathBuf, +} + +impl WorkerReadyProbe { + fn new() -> Self { + let path = soth_home_dir() + .join("run") + .join(format!("worker_ready_{}.json", uuid::Uuid::new_v4())); + Self { path } + } + + fn is_ready(&self) -> bool { + self.path.exists() + } +} + +impl Drop for WorkerReadyProbe { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +async fn spawn_proxy_process( + config_path: &Path, + listener_fd: Option, +) -> Result<(Child, WorkerReadyProbe)> { let current_exe = std::env::current_exe().context("resolve current executable for proxy worker")?; + let ready_probe = WorkerReadyProbe::new(); + if let Some(parent) = ready_probe.path.parent() { + let _ = std::fs::create_dir_all(parent); + } let mut cmd = Command::new(current_exe); // `start --daemon-child` + SOTH_PROXY_WORKER=1 selects the in-process MITM // runtime path in `run()` below, replacing the historical `soth-proxy` @@ -623,6 +657,10 @@ async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Re cmd.arg("start").arg("--daemon-child"); cmd.env(PROXY_WORKER_ENV, "1"); cmd.env("SOTH_PROXY_CONFIG", config_path); + cmd.env( + soth_proxy::drain_signal::WORKER_READY_FILE_ENV, + &ready_probe.path, + ); if let Some(fd) = listener_fd { cmd.env("SOTH_LISTENER_FD", fd.to_string()); } @@ -643,8 +681,10 @@ async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Re const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; cmd.creation_flags(CREATE_NO_WINDOW | CREATE_NEW_PROCESS_GROUP); } - cmd.spawn() - .map_err(|error| anyhow::anyhow!("failed launching proxy worker: {error}")) + let child = cmd + .spawn() + .map_err(|error| anyhow::anyhow!("failed launching proxy worker: {error}"))?; + Ok((child, ready_probe)) } /// Env var toggle that re-executed child processes use to enter in-process @@ -790,16 +830,56 @@ async fn supervise_proxy( } ProxyExit::Reload => { info!("reload requested (SIGHUP or network change) — performing graceful child rotation"); - let mut new_child = spawn_proxy_process(config_path, listener_fd) + // Windows: the port cannot be shared (no FD passing; the + // worker binds with SO_EXCLUSIVEADDRUSE), so rotation is + // sequenced — signal the drain FIRST. The old child closes + // its listener within milliseconds (freeing the port for + // the new child's bind-retry loop) and keeps draining + // in-flight flows in the background while the new child + // starts. On Unix the inherited-listener overlap flow is + // kept: new child up first, then drain the old. + #[cfg(windows)] + let old_child_draining = { + let signaled = child.id().map(signal_windows_drain_event).unwrap_or(false); + if !signaled { + warn!("could not signal drain event on old proxy child; hard-killing before respawn"); + let _ = terminate_child(child).await; + } + signaled + }; + + let (mut new_child, ready_probe) = spawn_proxy_process(config_path, listener_fd) .await .context("spawn new soth-proxy for graceful rotation")?; - if let Err(error) = wait_for_listener_start(&mut new_child, expected_port).await { + if let Err(error) = + wait_for_listener_start(&mut new_child, expected_port, &ready_probe).await + { + // Unix: the old child is untouched, keep serving on it. + // Windows: the old child is already draining; fall + // through — the supervisor loop observes its exit and + // respawns via the normal restart path. warn!(error = %error, "new proxy child failed to start; keeping old child"); let _ = terminate_child(&mut new_child).await; continue; } info!("new proxy child healthy — draining old child"); + #[cfg(unix)] graceful_stop_child(child).await?; + #[cfg(windows)] + if old_child_draining { + match tokio::time::timeout(Duration::from_secs(30), child.wait()).await { + Ok(Ok(status)) => { + info!(status = %status, "old proxy child exited after drain"); + } + Ok(Err(error)) => { + warn!(error = %error, "error waiting for old proxy child"); + } + Err(_) => { + warn!("old proxy child did not exit within 30s drain window; killing"); + let _ = terminate_child(child).await; + } + } + } *child = new_child; last_healthy = Instant::now(); consecutive_failures = 0; @@ -865,11 +945,17 @@ async fn supervise_proxy( ); tokio::time::sleep(Duration::from_millis(backoff_ms)).await; - *child = spawn_proxy_process(config_path, listener_fd) + let (respawned_child, ready_probe) = spawn_proxy_process(config_path, listener_fd) .await .context("respawn soth-proxy process")?; - if let Err(error) = - wait_for_listener_start_with_timeout(child, expected_port, next_startup_timeout).await + *child = respawned_child; + if let Err(error) = wait_for_listener_start_with_timeout( + child, + expected_port, + next_startup_timeout, + &ready_probe, + ) + .await { warn!( timeout_secs = next_startup_timeout.as_secs(), @@ -1036,18 +1122,42 @@ async fn graceful_stop_child(child: &mut Child) -> Result<()> { terminate_child(child).await } -async fn wait_for_listener_start(child: &mut Child, port: u16) -> Result<()> { - wait_for_listener_start_with_timeout(child, port, listener_startup_timeout()).await +async fn wait_for_listener_start( + child: &mut Child, + port: u16, + ready_probe: &WorkerReadyProbe, +) -> Result<()> { + wait_for_listener_start_with_timeout(child, port, listener_startup_timeout(), ready_probe).await } +/// Wait until the spawned worker reports ready. +/// +/// Primary signal: the worker's ready-file (written by that exact child once +/// its listener is live). The old TCP probe on `127.0.0.1:` could not +/// distinguish the new child's listener from the old child's during rotation +/// — and on Unix it didn't even prove the worker was up, since the +/// supervisor itself holds the inherited listener. The probe is kept only as +/// a late fallback for version-skewed workers that predate the ready-file +/// protocol (possible across hot-updates): it is consulted only after half +/// the startup budget has elapsed with no ready-file. async fn wait_for_listener_start_with_timeout( child: &mut Child, port: u16, timeout: Duration, + ready_probe: &WorkerReadyProbe, ) -> Result<()> { - let deadline = Instant::now() + timeout; + let start = Instant::now(); + let deadline = start + timeout; + let probe_fallback_after = start + timeout / 2; loop { - if is_local_listener_ready(port) { + if ready_probe.is_ready() { + return Ok(()); + } + if Instant::now() >= probe_fallback_after && is_local_listener_ready(port) { + warn!( + port, + "worker ready-file not seen but port answers; accepting via legacy probe (version-skewed worker?)" + ); return Ok(()); } if let Some(status) = child @@ -1059,7 +1169,7 @@ async fn wait_for_listener_start_with_timeout( if Instant::now() >= deadline { let _ = terminate_child(child).await; anyhow::bail!( - "soth-proxy did not open 127.0.0.1:{} within {}s startup timeout", + "soth-proxy did not report ready on 127.0.0.1:{} within {}s startup timeout", port, timeout.as_secs() ); @@ -1828,6 +1938,18 @@ async fn run_classify_daemon_worker() -> Result<()> { mod tests { use super::*; + #[test] + fn worker_ready_probe_removes_file_on_drop() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("worker_ready_test.json"); + let probe = WorkerReadyProbe { path: path.clone() }; + assert!(!probe.is_ready()); + std::fs::write(&path, b"{\"pid\": 1}\n").expect("write ready file"); + assert!(probe.is_ready()); + drop(probe); + assert!(!path.exists(), "probe drop must clean up the ready file"); + } + fn with_temp_home(f: impl FnOnce(std::path::PathBuf) -> T + std::panic::UnwindSafe) -> T { let guard = crate::commands::proxy::lock_test_env(); let temp = tempfile::tempdir().expect("tempdir"); diff --git a/crates/soth-proxy/src/drain_signal.rs b/crates/soth-proxy/src/drain_signal.rs index 189b727..effe61b 100644 --- a/crates/soth-proxy/src/drain_signal.rs +++ b/crates/soth-proxy/src/drain_signal.rs @@ -13,6 +13,41 @@ use tracing::info; +/// Env var carrying the supervisor-chosen path of the readiness file. The +/// worker writes `{"pid": }` there once its listener is live; the +/// supervisor waits for the file instead of TCP-probing the port. +/// +/// The old probe (`TcpStream::connect(127.0.0.1:)`) could not tell +/// the new child's listener from the old child's during rotation — on Unix +/// it didn't even prove the worker was up, since the supervisor itself +/// holds the inherited listener and the kernel completes connects from the +/// backlog. The ready-file is written by the worker process itself, so it +/// is unambiguous. +pub const WORKER_READY_FILE_ENV: &str = "SOTH_WORKER_READY_FILE"; + +/// Write the readiness file if the supervisor requested one (best-effort: +/// standalone/foreground runs have no supervisor and skip this). Written +/// atomically via temp-file + rename so the supervisor never reads a +/// partial file. +pub(crate) fn write_worker_ready_file() { + let Ok(path) = std::env::var(WORKER_READY_FILE_ENV) else { + return; + }; + let path = std::path::PathBuf::from(path); + let body = format!("{{\"pid\": {}}}\n", std::process::id()); + let temp = path.with_extension("tmp"); + let written = + std::fs::write(&temp, body.as_bytes()).and_then(|()| std::fs::rename(&temp, &path)); + match written { + Ok(()) => info!(path = %path.display(), "worker readiness file written"), + Err(error) => tracing::warn!( + path = %path.display(), + %error, + "failed writing worker readiness file; supervisor will fall back to port probe" + ), + } +} + /// The name of the drain event for a worker process id. Kept in sync with /// the supervisor side (`soth-cli/src/commands/proxy/start.rs`). #[cfg(windows)] @@ -80,8 +115,41 @@ pub(crate) async fn wait_for_shutdown_signal() { } } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ready_file_written_atomically_when_env_set() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("worker_ready_test.json"); + // Env var is process-global; nothing else in this test binary reads + // it, and we clean up immediately after. + std::env::set_var(WORKER_READY_FILE_ENV, &path); + write_worker_ready_file(); + std::env::remove_var(WORKER_READY_FILE_ENV); + + let body = std::fs::read_to_string(&path).expect("ready file written"); + assert!(body.contains(&format!("\"pid\": {}", std::process::id()))); + assert!( + !path.with_extension("tmp").exists(), + "temp file must be renamed away" + ); + } + + #[test] + fn ready_file_skipped_without_env() { + // Must not panic or create anything when unsupervised. + std::env::remove_var(WORKER_READY_FILE_ENV); + write_worker_ready_file(); + } +} + /// Create the named drain event and block until the supervisor sets it. /// Returns `true` when the event fired, `false` on any setup/wait failure. +// unsafe_code is denied crate-wide; this function is one of the two FFI +// exceptions (with sqlite_vec) — kernel event syscalls have no safe wrapper. +#[allow(unsafe_code)] #[cfg(windows)] fn wait_for_drain_event(pid: u32) -> bool { use windows_sys::Win32::Foundation::{CloseHandle, WAIT_OBJECT_0}; diff --git a/crates/soth-proxy/src/lib.rs b/crates/soth-proxy/src/lib.rs index dffc9ba..69d8a4f 100644 --- a/crates/soth-proxy/src/lib.rs +++ b/crates/soth-proxy/src/lib.rs @@ -1,6 +1,7 @@ -// unsafe_code is denied crate-wide; the single exception is sqlite_vec, which -// must call sqlite3_auto_extension via FFI. deny (not forbid) is used so that -// the per-module allow override on sqlite_vec is permitted. +// unsafe_code is denied crate-wide; the exceptions are sqlite_vec (must call +// sqlite3_auto_extension via FFI) and drain_signal's Windows kernel-event +// syscalls. deny (not forbid) is used so that per-item allow overrides are +// permitted. #![deny(unsafe_code)] #![allow(clippy::too_many_arguments)] diff --git a/crates/soth-proxy/src/runtime.rs b/crates/soth-proxy/src/runtime.rs index be26372..7e256a6 100644 --- a/crates/soth-proxy/src/runtime.rs +++ b/crates/soth-proxy/src/runtime.rs @@ -337,6 +337,7 @@ async fn run_inner(ext_registry: Option) -> Result<()> { None => proxy.start().await.context("start mitm proxy")?, }; + crate::drain_signal::write_worker_ready_file(); info!("soth-proxy started; press Ctrl+C to stop"); crate::drain_signal::wait_for_shutdown_signal().await; @@ -631,6 +632,7 @@ async fn run_inner() -> Result<()> { None => proxy.start().await.context("start mitm proxy")?, }; + crate::drain_signal::write_worker_ready_file(); info!("soth-proxy started; press Ctrl+C to stop"); crate::drain_signal::wait_for_shutdown_signal().await; From 461034ae2e9d8048d9306c2ab6b49fdc45752149 Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Fri, 3 Jul 2026 12:44:07 +0400 Subject: [PATCH 4/6] ci: test soth-proxy + soth-cli natively on Windows in release-ci MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-PR lanes are ubuntu-only and the existing test-windows job skipped exactly the two crates that carry Windows-divergent runtime code: soth-proxy (named drain event) and soth-cli (ipconfig network watcher, sequenced rotation, drain-event signaling). Add them so cfg(windows) API mistakes surface at the release gate — and via workflow_dispatch for branch validation. Co-Authored-By: Claude Fable 5 --- .github/workflows/release-ci.yml | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/.github/workflows/release-ci.yml b/.github/workflows/release-ci.yml index 7d8e6fc..7d7e35c 100644 --- a/.github/workflows/release-ci.yml +++ b/.github/workflows/release-ci.yml @@ -75,6 +75,16 @@ jobs: cargo test --lib -p soth-extensions cargo test --lib -p soth-parse cargo test --lib -p soth-code + # soth-proxy and soth-cli carry Windows-divergent runtime code the + # ubuntu PR lanes never compile: the named drain event + # (drain_signal), the ipconfig-based network watcher, and the + # sequenced rotation in the supervisor. Test them natively so + # cfg(windows) API mistakes surface before a release. + - name: proxy + cli tests on Windows + shell: bash + run: | + cargo test --lib -p soth-proxy + cargo test --bins -p soth-cli # Phase 4b sidecar updater — Windows cross-compile validation. # The crate is windows-only at runtime; we cross-compile from From ab110c03c6100668225d8a0844c308559a83326b Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Fri, 3 Jul 2026 12:51:14 +0400 Subject: [PATCH 5/6] =?UTF-8?q?ci:=20fix=20test-windows-sidecar=20?= =?UTF-8?q?=E2=80=94=20sidecar=20crate=20has=20no=20lib=20target?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cargo test --lib -p soth-cli-update-sidecar` fails with "no library targets found" (the crate is binary-only), so the job has been broken since it was added — masked because it only runs on tag push / workflow_dispatch. Test the bin target instead. Co-Authored-By: Claude Fable 5 --- .github/workflows/release-ci.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/release-ci.yml b/.github/workflows/release-ci.yml index 7d7e35c..5becd05 100644 --- a/.github/workflows/release-ci.yml +++ b/.github/workflows/release-ci.yml @@ -100,7 +100,9 @@ jobs: - shell: bash run: | cargo build -p soth-cli-update-sidecar --bin soth-update --release - cargo test --lib -p soth-cli-update-sidecar + # The sidecar is a binary-only crate: `cargo test --lib` errors + # with "no library targets found". Test the bin target instead. + cargo test --bins -p soth-cli-update-sidecar # FFI conformance is also a per-PR lane (paths-gated on # sdk-core / detect / classify / bindings / fixtures), but we From 77d8d7f045e50de4eecf2327bd112ef95f91d2fb Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Fri, 3 Jul 2026 13:02:08 +0400 Subject: [PATCH 6/6] test(cli): override USERPROFILE alongside HOME in temp-home helpers dirs::home_dir() reads USERPROFILE on Windows, not HOME, so the with_temp_home sandboxes leaked: `~` expansion (e.g. the default bundle_dir "~/.soth/bundle") resolved into the real runner profile and start_forwards_allow_daemon_child_fallback_flag failed on the new native Windows lane with "Bundle directory ... missing". Override and restore both vars in all three helpers (command_graph, proxy/start, proxy/system); harmless on Unix where USERPROFILE is unused. Co-Authored-By: Claude Fable 5 --- crates/soth-cli/src/command_graph.rs | 9 +++++++++ crates/soth-cli/src/commands/proxy/start.rs | 12 ++++++++++++ crates/soth-cli/src/commands/proxy/system.rs | 12 ++++++++++++ 3 files changed, 33 insertions(+) diff --git a/crates/soth-cli/src/command_graph.rs b/crates/soth-cli/src/command_graph.rs index 9be8908..14d1e3c 100644 --- a/crates/soth-cli/src/command_graph.rs +++ b/crates/soth-cli/src/command_graph.rs @@ -1510,9 +1510,14 @@ mod tests { std::fs::create_dir_all(&bundle_dir).expect("create bundle dir"); std::fs::write(bundle_dir.join("manifest.json"), "{}").expect("write bundle marker"); let old_home = env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — without + // overriding both, `~` expansion escapes the temp sandbox into the + // real user profile and the test fails on missing bundle state. + let old_userprofile = env::var_os("USERPROFILE"); let old_soth_home = env::var_os("SOTH_HOME_DIR"); unsafe { env::set_var("HOME", temp.path()); + env::set_var("USERPROFILE", temp.path()); env::set_var("SOTH_HOME_DIR", &soth_home); } let result = std::panic::catch_unwind(|| f(&temp)); @@ -1520,6 +1525,10 @@ mod tests { Some(value) => unsafe { env::set_var("HOME", value) }, None => unsafe { env::remove_var("HOME") }, } + match old_userprofile { + Some(value) => unsafe { env::set_var("USERPROFILE", value) }, + None => unsafe { env::remove_var("USERPROFILE") }, + } match old_soth_home { Some(value) => unsafe { env::set_var("SOTH_HOME_DIR", value) }, None => unsafe { env::remove_var("SOTH_HOME_DIR") }, diff --git a/crates/soth-cli/src/commands/proxy/start.rs b/crates/soth-cli/src/commands/proxy/start.rs index df63c58..0bd99f5 100644 --- a/crates/soth-cli/src/commands/proxy/start.rs +++ b/crates/soth-cli/src/commands/proxy/start.rs @@ -1954,10 +1954,14 @@ mod tests { let guard = crate::commands::proxy::lock_test_env(); let temp = tempfile::tempdir().expect("tempdir"); let old_home = std::env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — override + // both so `~` expansion can't escape the temp sandbox. + let old_userprofile = std::env::var_os("USERPROFILE"); let old_soth_home = std::env::var_os("SOTH_HOME_DIR"); let soth_home = temp.path().join(".soth"); unsafe { std::env::set_var("HOME", temp.path()); + std::env::set_var("USERPROFILE", temp.path()); std::env::set_var("SOTH_HOME_DIR", &soth_home); } @@ -1971,6 +1975,14 @@ mod tests { std::env::remove_var("HOME"); }, } + match old_userprofile { + Some(value) => unsafe { + std::env::set_var("USERPROFILE", value); + }, + None => unsafe { + std::env::remove_var("USERPROFILE"); + }, + } match old_soth_home { Some(value) => unsafe { std::env::set_var("SOTH_HOME_DIR", value); diff --git a/crates/soth-cli/src/commands/proxy/system.rs b/crates/soth-cli/src/commands/proxy/system.rs index 316ac94..f30716b 100644 --- a/crates/soth-cli/src/commands/proxy/system.rs +++ b/crates/soth-cli/src/commands/proxy/system.rs @@ -1695,10 +1695,14 @@ mod tests { let guard = crate::commands::proxy::lock_test_env(); let temp = tempfile::tempdir().expect("tempdir"); let old_home = env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — override + // both so `~` expansion can't escape the temp sandbox. + let old_userprofile = env::var_os("USERPROFILE"); let old_soth_home = env::var_os("SOTH_HOME_DIR"); let soth_home = temp.path().join(".soth"); unsafe { env::set_var("HOME", temp.path()); + env::set_var("USERPROFILE", temp.path()); env::set_var("SOTH_HOME_DIR", &soth_home); } let result = f(); @@ -1710,6 +1714,14 @@ mod tests { env::remove_var("HOME"); }, } + match old_userprofile { + Some(value) => unsafe { + env::set_var("USERPROFILE", value); + }, + None => unsafe { + env::remove_var("USERPROFILE"); + }, + } match old_soth_home { Some(value) => unsafe { env::set_var("SOTH_HOME_DIR", value);