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();