From b281e7c10bdb84d79b297042a920ecd00a0f725c Mon Sep 17 00:00:00 2001 From: cfal Date: Sat, 24 Oct 2026 22:52:07 +0000 Subject: [PATCH] fix(wireguard): separate lazy-group demand from carrier maintenance Suppress automatic-group activity during background association retries. Propagate new outbound TCP and UDP demand through shared and nested carriers without selecting or redialing their pinned routes. Cover idle failure retries and foreground versus background demand. --- src/client_proxy_chain.rs | 18 +++++ src/tcp/socket_connector.rs | 3 + src/wireguard/detour.rs | 131 +++++++++++++++++++++++++++++++++++- src/wireguard/mod.rs | 37 ++++++++-- 4 files changed, 182 insertions(+), 7 deletions(-) diff --git a/src/client_proxy_chain.rs b/src/client_proxy_chain.rs index ee9a3a8c..be3d536e 100644 --- a/src/client_proxy_chain.rs +++ b/src/client_proxy_chain.rs @@ -269,6 +269,24 @@ impl ClientProxyChain { } } + pub(crate) fn record_carrier_demand(&self) { + if crate::control::url_test::is_background() { + return; + } + // Shared carriers can have several pinned associations. Wake their groups + // without selecting a new route or attributing a destination failure to them. + for origin in self.origins.iter().flatten().flatten() { + for (group, _) in &origin.path { + origin.selection.automatic.touch(group, false); + } + } + for entry in &self.initial_hop { + if let InitialHopEntry::Wireguard(socket) = entry.as_ref() { + socket.record_demand(); + } + } + } + fn route_names(&self, indices: &[usize]) -> Vec { indices .iter() diff --git a/src/tcp/socket_connector.rs b/src/tcp/socket_connector.rs index effbd4f2..8070a335 100644 --- a/src/tcp/socket_connector.rs +++ b/src/tcp/socket_connector.rs @@ -34,6 +34,9 @@ use crate::resolver::Resolver; reason = "async-trait adds must_use to boxed futures" )] pub trait SocketConnector: Send + Sync + Debug { + /// Propagate new forwarding demand into a shared carrier without redialing it. + fn record_demand(&self) {} + async fn connect_quic( &self, _resolver: &Arc, diff --git a/src/wireguard/detour.rs b/src/wireguard/detour.rs index fd72ac73..74909e28 100644 --- a/src/wireguard/detour.rs +++ b/src/wireguard/detour.rs @@ -111,7 +111,7 @@ pub(super) fn start( incoming: incoming.clone(), drops: drops.clone(), }; - workers.spawn(worker.run()); + workers.spawn(crate::control::url_test::BACKGROUND.scope((), worker.run())); } drop(incoming); let shutdown = cancel.clone(); @@ -349,6 +349,135 @@ mod tests { .unwrap(); } + #[tokio::test(start_paused = true)] + async fn url_test_idle_reconnects_do_not_activate_lazy_groups() { + #[derive(Debug)] + struct FailingSocket(Arc); + + #[async_trait::async_trait] + impl SocketConnector for FailingSocket { + async fn connect( + &self, + _: &Arc, + _: &ResolvedLocation, + ) -> io::Result> { + unreachable!() + } + + async fn connect_udp_bidirectional( + &self, + _: &Arc, + _: ResolvedLocation, + ) -> io::Result> { + self.0.fetch_add(1, Ordering::Relaxed); + Err(io::ErrorKind::ConnectionRefused.into()) + } + + fn bind_interface(&self) -> Option<&str> { + None + } + } + + let mut selection = crate::control::selection::SelectionView::default(); + selection + .automatic + .insert( + "carrier".into(), + crate::config::UrlTestConfig { + url: "http://probe.invalid/".into(), + interval: 10, + timeout: 1000, + tolerance: 50, + lazy: true, + }, + vec!["peer".into()], + ) + .unwrap(); + let selection = Arc::new(selection); + let attempts = Arc::new(AtomicU64::new(0)); + let chain = Arc::new( + ClientProxyChain::new( + vec![InitialHopEntry::Direct(Arc::new(FailingSocket( + attempts.clone(), + )))], + vec![], + ) + .with_origins(vec![vec![Some(crate::control::catalog::MemberOrigin { + name: Arc::from("peer"), + path: vec![("carrier".into(), "peer".into())], + selection: selection.clone(), + })]]), + ); + let group = &selection.automatic.0["carrier"]; + assert!(group.due_members(Instant::now()).is_some()); + let (_sender, receiver, released) = start( + "idle".into(), + chain.clone(), + Arc::new(NativeResolver::new()), + ["127.0.0.1:51820".parse().unwrap()].into_iter(), + Arc::new(AtomicU64::new(0)), + ); + for _ in 0..20 { + tokio::time::advance(Duration::from_secs(1)).await; + tokio::task::yield_now().await; + } + assert!(attempts.load(Ordering::Relaxed) > 1); + assert!(group.due_members(Instant::now()).is_none()); + drop(receiver); + released.await.unwrap(); + + let stack = crate::ip_stack::driver::PacketStack::new( + vec!["10.0.0.2".parse().unwrap()], + 1420, + false, + Default::default(), + ) + .unwrap(); + let connector = |carrier| { + Arc::new(crate::wireguard::Connector { + name: "demand".into(), + active: Mutex::new(Some(crate::wireguard::ActiveEndpoint { + handle: stack.handle.clone(), + carrier: Some(carrier), + })), + addresses: vec!["10.0.0.2".parse().unwrap()], + routes: vec!["0.0.0.0/0".parse().unwrap()], + }) + }; + let outer = connector(chain); + let nested = connector(Arc::new(ClientProxyChain::new( + vec![InitialHopEntry::Wireguard(outer)], + vec![], + ))); + let resolver: Arc = Arc::new(NativeResolver::new()); + let target: ResolvedLocation = + NetLocation::from_ip_addr("10.0.0.1".parse().unwrap(), 80).into(); + for udp in [false, true] { + for background in [true, false] { + let connect = async { + if udp { + let attempt = nested.connect_udp_bidirectional(&resolver, target.clone()); + tokio::pin!(attempt); + let _ = futures::poll!(&mut attempt); + } else { + let attempt = nested.connect(&resolver, &target); + tokio::pin!(attempt); + let _ = futures::poll!(&mut attempt); + } + }; + if background { + crate::control::url_test::BACKGROUND + .scope((), connect) + .await; + } else { + connect.await; + } + assert_eq!(group.due_members(Instant::now()).is_some(), !background); + tokio::time::advance(Duration::from_secs(11)).await; + } + } + } + async fn read(stream: &mut dyn AsyncMessageStream) -> Vec { let mut bytes = vec![0; 65536]; let mut buffer = ReadBuf::new(&mut bytes); diff --git a/src/wireguard/mod.rs b/src/wireguard/mod.rs index d355bd23..40317ebc 100644 --- a/src/wireguard/mod.rs +++ b/src/wireguard/mod.rs @@ -44,10 +44,16 @@ fn canonical_socket_addr(mut address: SocketAddr) -> SocketAddr { struct Connector { name: String, - active: Mutex>, + active: Mutex>, addresses: Vec, routes: Vec, } + +#[derive(Clone)] +struct ActiveEndpoint { + handle: StackHandle, + carrier: Option>, +} impl fmt::Debug for Connector { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("WireguardConnector") @@ -57,12 +63,16 @@ impl fmt::Debug for Connector { } impl Connector { fn handle(&self) -> stdio::Result { - self.active.lock().clone().ok_or_else(|| { + let active = self.active.lock().clone().ok_or_else(|| { stdio::Error::new( stdio::ErrorKind::NotConnected, format!("WireGuard endpoint {} is not active", self.name), ) - }) + })?; + if let Some(carrier) = active.carrier { + carrier.record_carrier_demand(); + } + Ok(active.handle) } fn reachable(&self, address: SocketAddr) -> bool { self.addresses @@ -99,6 +109,17 @@ impl Connector { } #[async_trait] impl SocketConnector for Connector { + fn record_demand(&self) { + let carrier = self + .active + .lock() + .as_ref() + .and_then(|active| active.carrier.clone()); + if let Some(carrier) = carrier { + carrier.record_carrier_demand(); + } + } + async fn connect( &self, resolver: &Arc, @@ -415,11 +436,12 @@ impl PreparedEndpoint { )?; let resolver = self.outer_resolver.clone(); // Rebuilding after preflight gives each start fresh pool indices and handler lifetimes. - let (udp, released, peers) = if let Some(chain) = self.build_outer_chain()? { + let outer_chain = self.build_outer_chain()?; + let (udp, released, peers) = if let Some(chain) = &outer_chain { let peers = self.peers_for_socket(None)?; let (sender, receiver, released) = detour::start( config.id.clone(), - chain, + chain.clone(), resolver, peers.iter().map(|peer| peer.endpoint.unwrap()), stack.drops.clone(), @@ -454,7 +476,10 @@ impl PreparedEndpoint { .map_err(|error| { stdio::Error::other(format!("WireGuard device startup failed: {error}")) })?; - *self.connector.active.lock() = Some(stack.handle.clone()); + *self.connector.active.lock() = Some(ActiveEndpoint { + handle: stack.handle.clone(), + carrier: outer_chain, + }); let connector = self.connector.clone(); let name = config.id.clone(); let cancel = CancellationToken::new();