fix(icmp): open proxy sockets on demand and activate dynamic subnets (#2759)

* fix(icmp): open proxy sockets only when forwarding echo requests

Keep local no-TUN echo replies independent of raw socket permissions.
Create the socket and receiver on the first outward request, so newly
added proxy networks and exit-node traffic work without eager sockets.

Serialize concurrent initialization and socket publication with stop.
Roll back partial resources on failure or cancellation, and retain an
opened socket until shutdown for in-flight replies and later requests.

Cover local replies, dynamic CIDR updates, concurrent first requests,
exit-node forwarding, failed opens, cancellation and shutdown.

* fix(proxy): start forwarding after the first dynamic subnet update

A TUN instance started without proxy networks skips the packet proxy.
Runtime updates previously published new CIDRs without starting it, so
the first dynamically added subnet could not forward packets.

Reapply the existing startup predicate after configuration publication.
Run proxy startup on the instance executor so its sockets and tasks
survive a temporary configuration caller. Roll back partial startup on
cancellation as well as errors, allowing a subsequent update to retry.

Cover first-subnet forwarding, mapped CIDRs, removal and re-addition,
system forwarding, bind failure, cancellation and executor ownership.

easytier-core/src/gateway/proxy/icmp_proxy_service.rs

@@ -11,6 +11,7 @@ use tokio::{
     sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
     task::JoinSet,
 };
+use tokio_util::sync::CancellationToken;
 
 use crate::{
     packet::ZCPacket,
@@ -26,23 +27,6 @@ use super::{
     traits::{IcmpProxyRuntime, IcmpProxySocket, ProxyRuntimeError},
 };
 
-async fn start_icmp_runtime<R: IcmpProxyRuntime + 'static>(
-    runtime: &R,
-    no_tun: bool,
-) -> Result<Option<Arc<R::Socket>>, ProxyRuntimeError> {
-    match runtime.start_icmp().await {
-        Ok(socket) => Ok(Some(socket)),
-        Err(err) => {
-            runtime.stop_icmp();
-            if !no_tun {
-                return Err(err);
-            }
-            tracing::warn!(?err, "start ICMP runtime failed without TUN");
-            Ok(None)
-        }
-    }
-}
-
 pub struct IcmpProxyService<R: IcmpProxyRuntime + 'static> {
     peer_manager: Arc<PeerManagerCore>,
     runtime: Arc<R>,
@@ -52,6 +36,8 @@ pub struct IcmpProxyService<R: IcmpProxyRuntime + 'static> {
     pipeline_guard: std::sync::Mutex<Option<PipelineRegistrationGuard>>,
     tasks: std::sync::Mutex<JoinSet<()>>,
     socket: std::sync::Mutex<Option<Arc<R::Socket>>>,
+    socket_init: tokio::sync::Mutex<()>,
+    stopped: CancellationToken,
     runtime_started: AtomicBool,
     started: AtomicBool,
 }
@@ -73,6 +59,8 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
             pipeline_guard: std::sync::Mutex::new(None),
             tasks: std::sync::Mutex::new(JoinSet::new()),
             socket: std::sync::Mutex::new(None),
+            socket_init: tokio::sync::Mutex::new(()),
+            stopped: CancellationToken::new(),
             runtime_started: AtomicBool::new(false),
             started: AtomicBool::new(false),
         })
@@ -84,17 +72,6 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
         }
 
         let snapshot = self.runtime.proxy_runtime_snapshot();
-        let socket = match start_icmp_runtime(self.runtime.as_ref(), snapshot.no_tun).await {
-            Ok(socket) => socket,
-            Err(err) => {
-                self.started.store(false, Ordering::Release);
-                return Err(err);
-            }
-        };
-        let runtime_started = socket.is_some();
-        self.runtime_started
-            .store(runtime_started, Ordering::Release);
-        self.socket.lock().unwrap().clone_from(&socket);
 
         if let Some(mut response_rx) = self.response_rx.lock().unwrap().take() {
             let peer_manager = self.peer_manager.clone();
@@ -113,33 +90,6 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
             });
         }
 
-        if let Some(socket) = socket {
-            let service = Arc::downgrade(self);
-            self.tasks.lock().unwrap().spawn(async move {
-                loop {
-                    let recv_result = socket.recv().await;
-                    let (peer_ip, mut packet) = match recv_result {
-                        Ok(packet) => packet,
-                        Err(err) => {
-                            tracing::error!(?err, "receive ICMP packet failed");
-                            continue;
-                        }
-                    };
-                    if packet.is_empty() {
-                        tracing::error!("received empty ICMP packet");
-                        break;
-                    }
-                    let IpAddr::V4(peer_ip) = peer_ip else {
-                        continue;
-                    };
-                    let Some(service) = service.upgrade() else {
-                        break;
-                    };
-                    service.handle_socket_response(peer_ip, &mut packet);
-                }
-            });
-        }
-
         let service = Arc::downgrade(self);
         self.tasks.lock().unwrap().spawn(async move {
             loop {
@@ -176,9 +126,13 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
     }
 
     pub fn stop(&self) {
+        // Serialize shutdown with publication of a newly opened socket and its
+        // receive task. Opening itself must not hold this synchronous lock.
+        let mut socket = self.socket.lock().unwrap();
         if !self.started.swap(false, Ordering::AcqRel) {
             return;
         }
+        self.stopped.cancel();
         if let Some(guard) = self.pipeline_guard.lock().unwrap().take() {
             guard.close();
         }
@@ -186,7 +140,61 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
         if self.runtime_started.swap(false, Ordering::AcqRel) {
             self.runtime.stop_icmp();
         }
-        self.socket.lock().unwrap().take();
+        socket.take();
+    }
+
+    async fn socket(self: &Arc<Self>) -> Result<Arc<R::Socket>, ProxyRuntimeError> {
+        let _init = self.socket_init.lock().await;
+        if !self.started.load(Ordering::Acquire) || self.stopped.is_cancelled() {
+            return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into());
+        }
+        if let Some(socket) = self.socket.lock().unwrap().clone() {
+            return Ok(socket);
+        }
+
+        // Roll back partial resources on errors, caller cancellation or stop.
+        let rollback = guarden::guard! { self.runtime.stop_icmp(); };
+        let socket = tokio::select! {
+            biased;
+            _ = self.stopped.cancelled() => {
+                return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into());
+            }
+            socket = self.runtime.start_icmp() => socket?,
+        };
+        let mut current = self.socket.lock().unwrap();
+        if !self.started.load(Ordering::Acquire) {
+            return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into());
+        }
+
+        let service = Arc::downgrade(self);
+        let receiver = socket.clone();
+        self.tasks.lock().unwrap().spawn(async move {
+            loop {
+                let recv_result = receiver.recv().await;
+                let (peer_ip, mut packet) = match recv_result {
+                    Ok(packet) => packet,
+                    Err(err) => {
+                        tracing::error!(?err, "receive ICMP packet failed");
+                        continue;
+                    }
+                };
+                if packet.is_empty() {
+                    tracing::error!("received empty ICMP packet");
+                    break;
+                }
+                let IpAddr::V4(peer_ip) = peer_ip else {
+                    continue;
+                };
+                let Some(service) = service.upgrade() else {
+                    break;
+                };
+                service.handle_socket_response(peer_ip, &mut packet);
+            }
+        });
+        *current = Some(socket.clone());
+        self.runtime_started.store(true, Ordering::Release);
+        rollback.defuse();
+        Ok(socket)
     }
 
     async fn handle_peer_packet(self: Arc<Self>, packet: ZCPacket) -> Option<ZCPacket> {
@@ -204,14 +212,13 @@ impl<R: IcmpProxyRuntime + 'static> IcmpProxyService<R> {
                 destination,
                 packet: request,
             } => {
-                let socket = self.socket.lock().unwrap().clone();
-                match socket {
-                    Some(socket) => {
+                match self.socket().await {
+                    Ok(socket) => {
                         if let Err(err) = socket.send(destination, &request).await {
                             tracing::error!(?err, "send ICMP packet through runtime failed");
                         }
                     }
-                    None => tracing::error!("send ICMP packet without a runtime socket"),
+                    Err(err) => tracing::error!(?err, "start ICMP runtime failed"),
                 }
                 None
             }
@@ -259,25 +266,59 @@ impl<R: IcmpProxyRuntime + 'static> PeerPacketFilter for IcmpProxyServiceFilter<
 
 #[cfg(test)]
 mod tests {
-    use std::{
-        net::{IpAddr, Ipv4Addr},
-        sync::{
-            Arc,
-            atomic::{AtomicUsize, Ordering},
-        },
-    };
+    use std::sync::atomic::AtomicUsize;
+
+    use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, IpProtocol, Ipv4Packet};
+    use tokio::sync::Semaphore;
 
     use super::*;
-    use crate::gateway::proxy::traits::{IcmpProxySocket, ProxyRuntimeInfo, ProxyRuntimeSnapshot};
+    use crate::{
+        config::peers::PeerRuntimeSnapshot,
+        gateway::proxy::{
+            cidr_table::{ProxyCidrRule, ProxyCidrSnapshot},
+            traits::{ProxyRuntimeInfo, ProxyRuntimeSnapshot},
+        },
+        packet::PacketType,
+        peers::peer_manager::PortablePeerManagerConfig,
+    };
 
     #[derive(Default)]
-    struct PartialStartRuntime {
-        stop_count: AtomicUsize,
+    struct TestSocket {
+        sent: std::sync::Mutex<Vec<Ipv4Addr>>,
+        receives: AtomicUsize,
     }
 
-    impl ProxyRuntimeInfo for PartialStartRuntime {
+    #[async_trait::async_trait]
+    impl IcmpProxySocket for TestSocket {
+        async fn send(
+            &self,
+            destination: Ipv4Addr,
+            _packet: &[u8],
+        ) -> Result<(), ProxyRuntimeError> {
+            self.sent.lock().unwrap().push(destination);
+            Ok(())
+        }
+
+        async fn recv(&self) -> Result<(IpAddr, Vec<u8>), ProxyRuntimeError> {
+            self.receives.fetch_add(1, Ordering::AcqRel);
+            std::future::pending().await
+        }
+    }
+
+    #[derive(Default)]
+    struct TestRuntime {
+        snapshot: ProxyRuntimeSnapshot,
+        socket: Arc<TestSocket>,
+        starts: AtomicUsize,
+        stops: AtomicUsize,
+        active: AtomicBool,
+        fail: AtomicBool,
+        start_gate: Option<Semaphore>,
+    }
+
+    impl ProxyRuntimeInfo for TestRuntime {
         fn proxy_runtime_snapshot(&self) -> ProxyRuntimeSnapshot {
-            ProxyRuntimeSnapshot::default()
+            self.snapshot
         }
 
         fn is_ip_local_virtual_ip(&self, _ip: &IpAddr) -> bool {
@@ -285,53 +326,252 @@ mod tests {
         }
     }
 
-    struct NoopIcmpSocket;
-
     #[async_trait::async_trait]
-    impl IcmpProxySocket for NoopIcmpSocket {
-        async fn send(
-            &self,
-            _destination: Ipv4Addr,
-            _packet: &[u8],
-        ) -> Result<(), ProxyRuntimeError> {
-            Ok(())
-        }
-
-        async fn recv(&self) -> Result<(IpAddr, Vec<u8>), ProxyRuntimeError> {
-            Err(std::io::Error::other("unused test socket").into())
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl IcmpProxyRuntime for PartialStartRuntime {
-        type Socket = NoopIcmpSocket;
+    impl IcmpProxyRuntime for TestRuntime {
+        type Socket = TestSocket;
 
         async fn start_icmp(&self) -> Result<Arc<Self::Socket>, ProxyRuntimeError> {
-            Err(std::io::Error::other("partial start").into())
+            self.starts.fetch_add(1, Ordering::AcqRel);
+            self.active.store(true, Ordering::Release);
+            if let Some(gate) = &self.start_gate {
+                gate.acquire().await.unwrap().forget();
+            }
+            tokio::task::yield_now().await;
+            if self.fail.load(Ordering::Acquire) {
+                return Err(std::io::Error::other("partial start").into());
+            }
+            Ok(self.socket.clone())
         }
 
         fn stop_icmp(&self) {
-            self.stop_count.fetch_add(1, Ordering::AcqRel);
+            self.stops.fetch_add(1, Ordering::AcqRel);
+            self.active.store(false, Ordering::Release);
         }
     }
 
-    #[tokio::test]
-    async fn failed_runtime_start_rolls_back_partial_resources() {
-        let runtime = PartialStartRuntime::default();
+    fn service(
+        runtime: Arc<TestRuntime>,
+        table: Arc<ProxyCidrTable>,
+    ) -> Arc<IcmpProxyService<TestRuntime>> {
+        let (packet_tx, _packet_rx) = crate::host::packet::host_packet_channel();
+        let peer_manager = Arc::new(
+            PeerManagerCore::new_portable_for_test(
+                PortablePeerManagerConfig::new(PeerRuntimeSnapshot::default().runtime),
+                packet_tx,
+            )
+            .unwrap(),
+        );
+        IcmpProxyService::new(peer_manager, runtime, table, Duration::from_secs(10))
+    }
 
-        assert!(start_icmp_runtime(&runtime, false).await.is_err());
-        assert_eq!(runtime.stop_count.load(Ordering::Acquire), 1);
+    fn runtime(no_tun: bool, enable_exit_node: bool) -> Arc<TestRuntime> {
+        Arc::new(TestRuntime {
+            snapshot: ProxyRuntimeSnapshot {
+                virtual_ipv4: Some("10.0.0.1".parse().unwrap()),
+                no_tun,
+                enable_exit_node,
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+    }
+
+    fn echo_request(destination: &str) -> ZCPacket {
+        let mut bytes = [0_u8; 28];
+        let mut request = Icmpv4Packet::new_unchecked(&mut bytes[20..]);
+        request.set_msg_type(Icmpv4Message::EchoRequest);
+        request.set_echo_ident(7);
+        request.set_echo_seq_no(11);
+        request.fill_checksum();
+        let mut ipv4 = Ipv4Packet::new_unchecked(&mut bytes[..]);
+        ipv4.set_version(4);
+        ipv4.set_header_len(20);
+        ipv4.set_total_len(28);
+        ipv4.set_hop_limit(64);
+        ipv4.set_next_header(IpProtocol::Icmp);
+        ipv4.set_src_addr("10.0.0.2".parse().unwrap());
+        ipv4.set_dst_addr(destination.parse().unwrap());
+        ipv4.fill_checksum();
+        let mut packet = ZCPacket::new_with_payload(&bytes);
+        packet.fill_peer_manager_hdr(101, 202, PacketType::Data as u8);
+        packet
     }
 
     #[tokio::test]
-    async fn no_tun_suppresses_start_error_after_rollback() {
-        let runtime = PartialStartRuntime::default();
+    async fn no_tun_local_echo_does_not_open_socket() {
+        let runtime = runtime(true, false);
+        let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new()));
+        let mut responses = service.response_rx.lock().unwrap().take().unwrap();
+        service.start().await.unwrap();
+        assert!(
+            service
+                .clone()
+                .handle_peer_packet(echo_request("10.0.0.1"))
+                .await
+                .is_none()
+        );
+        let response = responses.try_recv().unwrap();
+        let ipv4 = Ipv4Packet::new_checked(response.payload()).unwrap();
+        let icmp = Icmpv4Packet::new_checked(ipv4.payload()).unwrap();
+        assert_eq!(icmp.msg_type(), Icmpv4Message::EchoReply);
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 0);
+        assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 0);
+        service.stop();
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 0);
+    }
 
-        let socket = start_icmp_runtime(&runtime, true).await.unwrap();
-        assert!(socket.is_none());
-        if socket.is_some() {
-            runtime.stop_icmp();
-        }
-        assert_eq!(runtime.stop_count.load(Ordering::Acquire), 1);
+    #[tokio::test]
+    async fn dynamic_proxy_networks_open_once_and_follow_updated_rules() {
+        let runtime = runtime(false, false);
+        let table = Arc::new(ProxyCidrTable::new());
+        let service = service(runtime.clone(), table.clone());
+        service.start().await.unwrap();
+        assert!(
+            service
+                .clone()
+                .handle_peer_packet(echo_request("192.0.2.42"))
+                .await
+                .is_some()
+        );
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 0);
+
+        table.update_snapshot(ProxyCidrSnapshot {
+            rules: vec![ProxyCidrRule {
+                cidr: "127.0.0.0/24".parse().unwrap(),
+                mapped_cidr: Some("192.0.2.0/24".parse().unwrap()),
+            }],
+        });
+        // The first concurrent requests must share both the socket and receiver.
+        futures::future::join_all((0..16).map(|_| {
+            service
+                .clone()
+                .handle_peer_packet(echo_request("192.0.2.42"))
+        }))
+        .await;
+        tokio::task::yield_now().await;
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 1);
+        assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 1);
+        assert_eq!(
+            *runtime.socket.sent.lock().unwrap(),
+            vec!["127.0.0.42".parse::<Ipv4Addr>().unwrap(); 16]
+        );
+
+        table.update_snapshot(ProxyCidrSnapshot::default());
+        assert!(
+            service
+                .clone()
+                .handle_peer_packet(echo_request("192.0.2.42"))
+                .await
+                .is_some()
+        );
+        assert_eq!(runtime.socket.sent.lock().unwrap().len(), 16);
+        // Keep the socket for in-flight replies and subsequent network additions.
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 0);
+        table.update_snapshot(ProxyCidrSnapshot {
+            rules: vec![ProxyCidrRule {
+                cidr: "192.0.2.0/24".parse().unwrap(),
+                mapped_cidr: None,
+            }],
+        });
+        assert!(
+            service
+                .clone()
+                .handle_peer_packet(echo_request("192.0.2.42"))
+                .await
+                .is_none()
+        );
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 1);
+        assert_eq!(
+            runtime.socket.sent.lock().unwrap().last(),
+            Some(&"192.0.2.42".parse().unwrap())
+        );
+        service.stop();
+        service.stop();
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 1);
+        assert!(!runtime.active.load(Ordering::Acquire));
+        assert!(service.socket().await.is_err());
+    }
+
+    #[tokio::test]
+    async fn exit_node_opens_socket_without_proxy_networks() {
+        let runtime = runtime(false, true);
+        let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new()));
+        service.start().await.unwrap();
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 0);
+        let mut packet = echo_request("192.0.2.42");
+        packet
+            .mut_peer_manager_header()
+            .unwrap()
+            .set_exit_node(true);
+        assert!(service.clone().handle_peer_packet(packet).await.is_none());
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 1);
+        assert_eq!(
+            *runtime.socket.sent.lock().unwrap(),
+            ["192.0.2.42".parse::<Ipv4Addr>().unwrap()]
+        );
+    }
+
+    #[tokio::test]
+    async fn failed_runtime_start_rolls_back_and_can_retry() {
+        let runtime = runtime(true, false);
+        runtime.fail.store(true, Ordering::Release);
+        let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new()));
+        let mut responses = service.response_rx.lock().unwrap().take().unwrap();
+        service.start().await.unwrap();
+        assert!(service.socket().await.is_err());
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 1);
+        assert!(!runtime.active.load(Ordering::Acquire));
+        assert!(
+            service
+                .clone()
+                .handle_peer_packet(echo_request("10.0.0.1"))
+                .await
+                .is_none()
+        );
+        assert!(responses.try_recv().is_ok());
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 1);
+        runtime.fail.store(false, Ordering::Release);
+        assert!(service.socket().await.is_ok());
+        service.stop();
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 2);
+    }
+
+    #[tokio::test]
+    async fn stop_cancels_pending_socket_start() {
+        let runtime = Arc::new(TestRuntime {
+            start_gate: Some(Semaphore::new(0)),
+            ..Default::default()
+        });
+        let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new()));
+        service.start().await.unwrap();
+        let mut opening = Box::pin(service.socket());
+        assert!(futures::poll!(&mut opening).is_pending());
+        assert!(runtime.active.load(Ordering::Acquire));
+        service.stop();
+        assert!(opening.await.is_err());
+        assert!(!runtime.active.load(Ordering::Acquire));
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 1);
+        assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 0);
+        assert!(service.socket().await.is_err());
+        assert_eq!(runtime.starts.load(Ordering::Acquire), 1);
+    }
+
+    #[tokio::test]
+    async fn cancelled_caller_rolls_back_socket_start() {
+        let runtime = Arc::new(TestRuntime {
+            start_gate: Some(Semaphore::new(0)),
+            ..Default::default()
+        });
+        let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new()));
+        service.start().await.unwrap();
+        let mut opening = Box::pin(service.socket());
+        assert!(futures::poll!(&mut opening).is_pending());
+        drop(opening);
+        assert!(!runtime.active.load(Ordering::Acquire));
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 1);
+        runtime.start_gate.as_ref().unwrap().add_permits(1);
+        assert!(service.socket().await.is_ok());
+        service.stop();
+        assert_eq!(runtime.stops.load(Ordering::Acquire), 2);
     }
 }

easytier-core/src/gateway/proxy/service.rs

@@ -379,12 +379,12 @@ where
             return Ok(());
         }
 
+        // Runtime updates can be cancelled while a host operation is pending.
+        // Roll back the startup flags and resources so a later update can retry.
+        let rollback = guarden::guard! { self.stop_started(); };
         self.runtime.latch_smoltcp();
         self.tcp_started.store(true, Ordering::Release);
-        if let Err(error) = self.tcp.start(true).await {
-            self.stop_started();
-            return Err(error);
-        }
+        self.tcp.start(true).await?;
 
         if let Some(icmp) = &self.icmp {
             self.icmp_started.store(true, Ordering::Release);
@@ -398,7 +398,6 @@ where
                     .proxy
                     .icmp_failure_is_fatal
                 {
-                    self.stop_started();
                     return Err(error);
                 }
                 tracing::warn!(?error, "optional ICMP proxy runtime failed to start");
@@ -407,6 +406,7 @@ where
 
         self.udp_started.store(true, Ordering::Release);
         self.udp.start().await;
+        rollback.defuse();
         Ok(())
     }
 

easytier-core/src/host/testkit.rs

@@ -133,6 +133,8 @@ pub struct TestHost {
         Option<tokio::sync::mpsc::UnboundedSender<(SocketAddr, tokio::io::DuplexStream)>>,
     pub tcp_listeners: Option<Arc<Mutex<TestTcpListeners>>>,
     pub reject_tcp_bind: Mutex<Option<SocketAddr>>,
+    pub tcp_bind_gate: Option<Arc<tokio::sync::Semaphore>>,
+    pub tcp_bind_runtime: Mutex<Option<tokio::runtime::Id>>,
     pub reject_socks5_listener: bool,
 }
 
@@ -179,6 +181,10 @@ impl VirtualTcpListenerFactory for TestHost {
 
     async fn bind_tcp(&self, options: TcpListenOptions) -> anyhow::Result<Arc<Self::Listener>> {
         self.tcp_binds.fetch_add(1, Ordering::Relaxed);
+        *self.tcp_bind_runtime.lock().unwrap() = Some(tokio::runtime::Handle::current().id());
+        if let Some(gate) = &self.tcp_bind_gate {
+            gate.acquire().await.unwrap().forget();
+        }
         if self.reject_socks5_listener && options.purpose == TcpListenPurpose::Socks5 {
             anyhow::bail!("test host rejected SOCKS5 listener");
         }

easytier-core/src/instance/lifecycle.rs

@@ -82,14 +82,23 @@ where
     }
 
     #[cfg(feature = "proxy-packet")]
-    async fn start_packet_proxy(&self) -> anyhow::Result<()> {
+    pub(super) async fn start_packet_proxy(&self) -> anyhow::Result<()> {
         let config = self.runtime_config.snapshot();
         let has_proxy_networks = !config.peer.runtime.core.routes.proxy_networks.is_empty();
         if config.services.proxy.should_start(has_proxy_networks) {
-            self.packet_proxy
-                .start()
+            if let Some(runtime) = self.runtime_handle() {
+                let proxy = self.packet_proxy.clone();
+                tokio_util::task::AbortOnDropHandle::new(
+                    runtime.spawn(async move { proxy.start().await }),
+                )
                 .await
-                .map_err(anyhow::Error::new)?;
+                .map_err(|error| anyhow::anyhow!("packet proxy start task failed: {error}"))??;
+            } else {
+                self.packet_proxy
+                    .start()
+                    .await
+                    .map_err(anyhow::Error::new)?;
+            }
         }
         Ok(())
     }

easytier-core/src/instance/mod.rs

@@ -988,6 +988,12 @@ where
         if let Some(cleanup) = cleanup {
             let _ = cleanup.await;
         }
+        #[cfg(feature = "proxy-packet")]
+        if self.state() == CoreInstanceState::Running && self.startup_plan.packet_proxy {
+            // A running instance may gain its first proxy network here. Start
+            // from the published configuration, just as during initial startup.
+            self.start_packet_proxy().await?;
+        }
         Ok(())
     }
 

easytier-core/src/instance/tests.rs

@@ -3487,6 +3487,353 @@ virtual_ip = "10.82.0.2/24"
         instance.stop().await;
     }
 
+    #[cfg(feature = "proxy-packet")]
+    mod dynamic_packet_proxy {
+        use std::net::Ipv4Addr;
+
+        use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, IpProtocol, Ipv4Packet};
+
+        use super::*;
+        use crate::{
+            gateway::proxy::icmp_host::{IcmpProxyHost, IcmpProxySocket, ProxyRuntimeError},
+            packet::{PacketType, ZCPacket},
+        };
+
+        struct RecordingIcmpHost {
+            opens: AtomicUsize,
+            requests: tokio::sync::mpsc::UnboundedSender<Ipv4Addr>,
+        }
+
+        #[async_trait]
+        impl IcmpProxyHost for RecordingIcmpHost {
+            async fn open_icmp_v4(
+                &self,
+                _context: SocketContext,
+            ) -> Result<Arc<dyn IcmpProxySocket>, ProxyRuntimeError> {
+                self.opens.fetch_add(1, Ordering::Relaxed);
+                Ok(Arc::new(self.requests.clone()))
+            }
+        }
+
+        #[async_trait]
+        impl IcmpProxySocket for tokio::sync::mpsc::UnboundedSender<Ipv4Addr> {
+            async fn send(
+                &self,
+                destination: Ipv4Addr,
+                _packet: &[u8],
+            ) -> Result<(), ProxyRuntimeError> {
+                self.send(destination).unwrap();
+                Ok(())
+            }
+
+            async fn recv(&self) -> Result<(IpAddr, Vec<u8>), ProxyRuntimeError> {
+                std::future::pending().await
+            }
+        }
+
+        fn echo_request(source_peer: u32, destination_peer: u32) -> ZCPacket {
+            let mut bytes = [0_u8; 28];
+            let mut icmp = Icmpv4Packet::new_unchecked(&mut bytes[20..]);
+            icmp.set_msg_type(Icmpv4Message::EchoRequest);
+            icmp.set_echo_ident(7);
+            icmp.set_echo_seq_no(1);
+            icmp.fill_checksum();
+            let mut ipv4 = Ipv4Packet::new_unchecked(&mut bytes[..]);
+            ipv4.set_version(4);
+            ipv4.set_header_len(20);
+            ipv4.set_total_len(28);
+            ipv4.set_hop_limit(64);
+            ipv4.set_next_header(IpProtocol::Icmp);
+            ipv4.set_src_addr("10.0.0.2".parse().unwrap());
+            ipv4.set_dst_addr("198.51.100.42".parse().unwrap());
+            ipv4.fill_checksum();
+            let mut packet = ZCPacket::new_with_payload(&bytes);
+            packet.fill_peer_manager_hdr(source_peer, destination_peer, PacketType::Data as u8);
+            packet
+        }
+
+        #[tokio::test]
+        async fn first_dynamic_proxy_network_starts_forwarding_and_reuses_resources() {
+            let mut config = test_config("dynamic-packet-proxy");
+            config.peer.snapshot.runtime.core.routes.ipv4 = Some(IpPrefix {
+                address: "10.0.0.1".parse().unwrap(),
+                prefix_len: 24,
+            });
+            let mut updated = runtime_snapshot(&config);
+            let host = Arc::new(TestHost::default());
+            let (packet_sink, mut packets) = tokio::sync::mpsc::channel(16);
+            let (requests, mut destinations) = tokio::sync::mpsc::unbounded_channel();
+            let icmp_host = Arc::new(RecordingIcmpHost {
+                opens: AtomicUsize::new(0),
+                requests,
+            });
+            let mut adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink));
+            adapters.icmp_proxy_host = Some(icmp_host.clone());
+            let instance = CoreInstance::new(config.clone(), adapters).unwrap();
+            config
+                .peer
+                .snapshot
+                .runtime
+                .core
+                .routes
+                .ipv4
+                .as_mut()
+                .unwrap()
+                .address = "10.0.0.2".parse().unwrap();
+            let source = build_instance(config).unwrap();
+            instance.start().await.unwrap();
+            source.start().await.unwrap();
+            assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0);
+            assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0);
+
+            let (client, server) = crate::tunnel::ring::create_ring_tunnel_pair();
+            let (client_result, server_result) = tokio::join!(
+                source.peer_manager.add_client_tunnel(client),
+                instance.peer_manager.add_tunnel_as_server(server),
+            );
+            client_result.unwrap();
+            server_result.unwrap();
+
+            let send = || async {
+                source
+                    .peer_manager
+                    .send_msg_for_proxy(
+                        echo_request(source.peer_id(), instance.peer_id()),
+                        instance.peer_id(),
+                    )
+                    .await
+                    .unwrap();
+            };
+            send().await;
+            tokio::time::timeout(Duration::from_secs(5), packets.recv())
+                .await
+                .unwrap()
+                .unwrap();
+            assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0);
+
+            Arc::make_mut(&mut updated.peer)
+                .runtime
+                .core
+                .routes
+                .proxy_networks = vec![proxy_network("192.0.2.0/24", Some("198.51.100.0/24"))];
+            instance
+                .update_runtime_config(updated.clone())
+                .await
+                .unwrap();
+            assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 1);
+            assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0);
+            send().await;
+            assert_eq!(
+                tokio::time::timeout(Duration::from_secs(5), destinations.recv())
+                    .await
+                    .unwrap(),
+                Some("192.0.2.42".parse().unwrap()),
+            );
+
+            let mut removed = updated.clone();
+            Arc::make_mut(&mut removed.peer)
+                .runtime
+                .core
+                .routes
+                .proxy_networks
+                .clear();
+            instance.update_runtime_config(removed).await.unwrap();
+            send().await;
+            tokio::time::timeout(Duration::from_secs(5), packets.recv())
+                .await
+                .unwrap()
+                .unwrap();
+            assert!(destinations.try_recv().is_err());
+
+            instance.update_runtime_config(updated).await.unwrap();
+            send().await;
+            assert_eq!(
+                tokio::time::timeout(Duration::from_secs(5), destinations.recv())
+                    .await
+                    .unwrap(),
+                Some("192.0.2.42".parse().unwrap()),
+            );
+            assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 1);
+            assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 1);
+            source.stop().await;
+            instance.stop().await;
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+        }
+
+        #[tokio::test]
+        async fn dynamic_proxy_start_respects_instance_state_and_system_forwarding() {
+            for forward_by_system in [false, true] {
+                let mut config = test_config("dynamic-proxy-start-policy");
+                config.connectivity.runtime.proxy.forward_by_system = forward_by_system;
+                let mut updated = runtime_snapshot(&config);
+                let host = Arc::new(TestHost::default());
+                let (packet_sink, _packets) = tokio::sync::mpsc::channel(16);
+                let instance = CoreInstance::new(
+                    config,
+                    adapters_with_host(host.clone(), None, Arc::new(packet_sink)),
+                )
+                .unwrap();
+                Arc::make_mut(&mut updated.peer)
+                    .runtime
+                    .core
+                    .routes
+                    .proxy_networks = vec![proxy_network("192.0.2.0/24", None)];
+                instance
+                    .update_runtime_config(updated.clone())
+                    .await
+                    .unwrap();
+                assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0);
+                instance.start().await.unwrap();
+                instance.update_runtime_config(updated).await.unwrap();
+                assert_eq!(
+                    host.tcp_binds.load(Ordering::Relaxed),
+                    usize::from(!forward_by_system),
+                );
+                instance.stop().await;
+            }
+        }
+
+        #[tokio::test]
+        async fn dynamic_proxy_start_reports_bind_failure_and_allows_retry() {
+            let config = test_config("dynamic-proxy-start-failure");
+            let mut updated = runtime_snapshot(&config);
+            let host = Arc::new(TestHost::default());
+            *host.reject_tcp_bind.lock().unwrap() = Some("0.0.0.0:20000".parse().unwrap());
+            let (packet_sink, _packets) = tokio::sync::mpsc::channel(16);
+            let instance = CoreInstance::new(
+                config,
+                adapters_with_host(host.clone(), None, Arc::new(packet_sink)),
+            )
+            .unwrap();
+            instance.start().await.unwrap();
+            Arc::make_mut(&mut updated.peer)
+                .runtime
+                .core
+                .routes
+                .proxy_networks = vec![proxy_network("192.0.2.0/24", None)];
+            let error = instance
+                .update_runtime_config(updated.clone())
+                .await
+                .unwrap_err();
+            assert!(error.to_string().contains("test host rejected TCP bind"));
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+            // Configuration publication precedes resource startup. Retrying the
+            // same update must activate the already published proxy network.
+            assert_eq!(instance.node_snapshot().await.proxy_networks.len(), 1);
+            *host.reject_tcp_bind.lock().unwrap() = None;
+            instance.update_runtime_config(updated).await.unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            instance.stop().await;
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+        }
+
+        #[tokio::test]
+        async fn cancelled_dynamic_proxy_start_allows_retry() {
+            let config = test_config("dynamic-proxy-start-cancelled");
+            let mut updated = runtime_snapshot(&config);
+            let gate = Arc::new(tokio::sync::Semaphore::new(0));
+            let host = Arc::new(TestHost {
+                tcp_bind_gate: Some(gate.clone()),
+                ..Default::default()
+            });
+            let (packet_sink, _packets) = tokio::sync::mpsc::channel(16);
+            let instance = CoreInstance::new(
+                config,
+                adapters_with_host(host.clone(), None, Arc::new(packet_sink)),
+            )
+            .unwrap();
+            instance.start().await.unwrap();
+            Arc::make_mut(&mut updated.peer)
+                .runtime
+                .core
+                .routes
+                .proxy_networks = vec![proxy_network("192.0.2.0/24", None)];
+            let update = tokio::spawn({
+                let instance = instance.clone();
+                let updated = updated.clone();
+                async move { instance.update_runtime_config(updated).await }
+            });
+            tokio::time::timeout(Duration::from_secs(5), async {
+                while host.tcp_binds.load(Ordering::Relaxed) == 0 {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .unwrap();
+            update.abort();
+            assert!(update.await.unwrap_err().is_cancelled());
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+
+            gate.add_permits(1);
+            instance.update_runtime_config(updated).await.unwrap();
+            assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 2);
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            instance.stop().await;
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+        }
+
+        #[test]
+        fn dynamic_proxy_start_uses_instance_executor_after_caller_runtime_exits() {
+            struct ExecutorRuntimeHost(tokio::runtime::Handle);
+
+            #[async_trait]
+            impl InstanceRuntimeHost for ExecutorRuntimeHost {
+                fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+                    Some(self.0.clone())
+                }
+
+                async fn prepare(
+                    &self,
+                    _packet_plane: Arc<CorePacketPlane>,
+                ) -> anyhow::Result<Option<Arc<dyn DhcpHost>>> {
+                    Ok(None)
+                }
+
+                async fn shutdown(&self) {}
+            }
+
+            let executor = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(2)
+                .enable_all()
+                .build()
+                .unwrap();
+            let caller = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            let host = Arc::new(TestHost::default());
+            let config = test_config("dynamic-proxy-executor");
+            let mut updated = runtime_snapshot(&config);
+            let instance = {
+                let _runtime = executor.enter();
+                let (packet_sink, _packets) = tokio::sync::mpsc::channel(16);
+                let mut adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink));
+                adapters.instance_runtime =
+                    Arc::new(ExecutorRuntimeHost(executor.handle().clone()));
+                CoreInstance::new(config, adapters).unwrap()
+            };
+            caller.block_on(async {
+                instance.start().await.unwrap();
+                assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0);
+                Arc::make_mut(&mut updated.peer)
+                    .runtime
+                    .core
+                    .routes
+                    .proxy_networks = vec![proxy_network("192.0.2.0/24", None)];
+                instance.update_runtime_config(updated).await.unwrap();
+            });
+            drop(caller);
+
+            assert_eq!(
+                *host.tcp_bind_runtime.lock().unwrap(),
+                Some(executor.handle().id()),
+            );
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            executor.block_on(instance.stop());
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+        }
+    }
+
     #[cfg(feature = "dhcp-ipv4")]
     #[tokio::test]
     async fn runtime_core_accepts_explicit_dhcp_runtime_snapshot() {