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

@@ -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<String> {
         indices
             .iter()

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<dyn Resolver>,

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<AtomicU64>);
+
+        #[async_trait::async_trait]
+        impl SocketConnector for FailingSocket {
+            async fn connect(
+                &self,
+                _: &Arc<dyn Resolver>,
+                _: &ResolvedLocation,
+            ) -> io::Result<Box<dyn AsyncStream>> {
+                unreachable!()
+            }
+
+            async fn connect_udp_bidirectional(
+                &self,
+                _: &Arc<dyn Resolver>,
+                _: ResolvedLocation,
+            ) -> io::Result<Box<dyn AsyncMessageStream>> {
+                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<dyn Resolver> = 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<u8> {
         let mut bytes = vec![0; 65536];
         let mut buffer = ReadBuf::new(&mut bytes);

src/wireguard/mod.rs

@@ -44,10 +44,16 @@ fn canonical_socket_addr(mut address: SocketAddr) -> SocketAddr {
 
 struct Connector {
     name: String,
-    active: Mutex<Option<StackHandle>>,
+    active: Mutex<Option<ActiveEndpoint>>,
     addresses: Vec<IpAddr>,
     routes: Vec<IpNetwork>,
 }
+
+#[derive(Clone)]
+struct ActiveEndpoint {
+    handle: StackHandle,
+    carrier: Option<Arc<crate::client_proxy_chain::ClientProxyChain>>,
+}
 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<StackHandle> {
-        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<dyn Resolver>,
@@ -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();