fix(socket): keep port forwards usable after interface recreation

Bind TCP and UDP port-forward listeners to their configured address
instead of implicitly pinning them to the current network interface.
An interface recreated with the same name and IP gets a new ifindex,
leaving device-bound listeners unable to receive traffic while accept
continues waiting. Keep explicit device-binding options unchanged.

Add TCP and UDP three-node regressions that recreate a TUN interface
without restarting the forwarding instance, plus device-option checks.

Fixes #2714.

.github/workflows/ohos.yml

@@ -138,7 +138,8 @@ jobs:
           set -euo pipefail
           cargo_version=$(cargo metadata --format-version 1 --no-deps \
             --manifest-path easytier/Cargo.toml | jq -r '.packages[0].version')
-          last_tag=$(git describe --tags --abbrev=0 HEAD 2>/dev/null || true)
+          # Rolling tags such as nightly are not package versions.
+          last_tag=$(git describe --tags --match 'v[0-9]*' --abbrev=0 HEAD 2>/dev/null || true)
           if [ -n "$last_tag" ]; then
             base_version=$(printf '%s\n' "$cargo_version" "${last_tag#v}" \
               | sort -V | tail -n 1)

docs/core-architecture.md

@@ -552,6 +552,13 @@ optional `tracing` feature selects the tokio-console subscriber integration;
 only that diagnostic profile pulls the main crate's `tracing-subscriber` and
 `console-subscriber` dependencies.
 
+The native CLI enables the `CORE` and `easytier_core::instance::manager` console
+targets by default. Instance startup failures are reported immediately with the
+instance ID and complete error chain, including the listener URL for bind
+failures. When all instances stop, the CLI's failure summary retains each failed
+instance's ID and saved error. Explicit console levels and `RUST_LOG` directives
+still control which records are emitted.
+
 Contrib applications and platform integrations may have independent logging
 requirements and are not implicitly wired to the native process logger.
 

easytier-core/src/gateway/mod.rs

@@ -27,6 +27,6 @@ pub use dataplane::{
     DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket,
 };
 #[cfg(feature = "proxy-smoltcp-stack")]
-pub(crate) use port_forward::PortForwardAdapter;
+pub(crate) use port_forward::{PortForwardAdapter, PreparedPortForwards};
 #[cfg(feature = "proxy-smoltcp-stack")]
 pub(crate) use socks5::Socks5GatewayAdapter;

easytier-core/src/gateway/port_forward.rs

@@ -1,6 +1,7 @@
 //! Host port-forward adapter backed by the data-plane runtime.
 
 use std::{
+    collections::HashMap,
     net::SocketAddr,
     sync::{
         Arc,
@@ -15,11 +16,11 @@ use quanta::Instant;
 use tokio::{
     select,
     sync::{Mutex, OwnedMutexGuard, OwnedSemaphorePermit, Semaphore},
-    task::JoinSet,
+    task::{JoinHandle, JoinSet},
 };
 use tokio_util::{
-    sync::{CancellationToken, DropGuard},
-    task::AbortOnDropHandle,
+    sync::CancellationToken,
+    task::{AbortOnDropHandle, TaskTracker},
 };
 
 use crate::{
@@ -94,11 +95,119 @@ struct UdpClientReservation {
     admission: OwnedMutexGuard<()>,
 }
 
+struct PortForwardListener {
+    config: Arc<std::sync::Mutex<PortForwardConfig>>,
+    cancel: CancellationToken,
+    task: AbortOnDropHandle<()>,
+    udp_tasks: TaskTracker,
+}
+
+enum PortForwardSocket<H: VirtualTcpListenerFactory + VirtualUdpSocketFactory> {
+    Tcp(Arc<H::Listener>),
+    Udp(Arc<<H as VirtualUdpSocketFactory>::Socket>),
+}
+
+/// Bound additions stay private until persistence and validation have succeeded.
+/// Dropping a prepared update leaves every existing listener and target intact.
+pub(crate) struct PreparedPortForwards<H>
+where
+    H: VirtualTcpSocketFactory + VirtualTcpListenerFactory + VirtualUdpSocketFactory,
+{
+    adapter: Arc<PortForwardAdapter<H>>,
+    configs: HashMap<(String, SocketAddr), PortForwardConfig>,
+    sockets: Vec<(PortForwardConfig, PortForwardSocket<H>)>,
+    consumer_lease: Option<DataPlaneConsumerLease>,
+    // Release prepared sockets before allowing another update to bind them.
+    operation: OwnedMutexGuard<()>,
+}
+
+impl<H> PreparedPortForwards<H>
+where
+    H: VirtualTcpSocketFactory + VirtualTcpListenerFactory + VirtualUdpSocketFactory,
+{
+    /// All fallible work is complete. Publish targets and start the bound sockets
+    /// without yielding, then retain the operation lock until removals release
+    /// their sockets, even if the caller stops waiting for this cleanup.
+    pub(crate) fn commit(self) -> JoinHandle<()> {
+        let Self {
+            adapter,
+            operation,
+            configs,
+            sockets,
+            consumer_lease,
+        } = self;
+        let _runtime = adapter.runtime_handle.as_ref().map(|handle| handle.enter());
+        let mut listeners = adapter.listeners.lock().unwrap();
+        let mut removed = Vec::new();
+        let mut response_tasks = Vec::new();
+        let removed_keys = listeners
+            .keys()
+            .filter(|key| !configs.contains_key(*key))
+            .cloned()
+            .collect::<Vec<_>>();
+        for key in removed_keys {
+            let listener = listeners.remove(&key).unwrap();
+            listener.cancel.cancel();
+            listener.task.abort();
+            {
+                let current = listener.config.lock().unwrap();
+                tracing::info!(?current, "port-forward removed by runtime config reload");
+                adapter.remove_udp_clients(&current, &mut response_tasks);
+            }
+            removed.push(listener);
+        }
+        for (key, listener) in listeners.iter() {
+            let mut current = listener.config.lock().unwrap();
+            if let Some(next) = configs.get(key)
+                && *current != *next
+            {
+                adapter.remove_udp_clients(&current, &mut response_tasks);
+                *current = next.clone();
+                adapter
+                    .events
+                    .emit(CoreEvent::GatewayPortForwardAdded(next.clone()));
+            }
+        }
+        for (cfg, socket) in sockets {
+            let key = (cfg.proto.clone(), cfg.bind_addr);
+            let listener = match socket {
+                PortForwardSocket::Tcp(socket) => adapter.add_tcp_port_forward(&cfg, socket),
+                PortForwardSocket::Udp(socket) => adapter.add_udp_port_forward(&cfg, socket),
+            };
+            listeners.insert(key, listener);
+            adapter.events.emit(CoreEvent::GatewayPortForwardAdded(cfg));
+        }
+        drop(listeners);
+        let mut lease = adapter.consumer_lease.lock().unwrap();
+        if let Some(consumer_lease) = consumer_lease {
+            lease.replace(consumer_lease);
+        }
+        let released_lease = if configs.is_empty() {
+            lease.take()
+        } else {
+            None
+        };
+        drop(lease);
+        tokio::spawn(async move {
+            let _operation = operation;
+            for listener in removed {
+                let _ = listener.task.await;
+                listener.udp_tasks.close();
+                listener.udp_tasks.wait().await;
+            }
+            for task in response_tasks {
+                let _ = task.await;
+            }
+            drop(released_lease);
+        })
+    }
+}
+
 pub(crate) struct PortForwardAdapter<H>
 where
     H: VirtualTcpSocketFactory + VirtualTcpListenerFactory + VirtualUdpSocketFactory,
 {
-    operation: Mutex<()>,
+    operation: Arc<Mutex<()>>,
     started: AtomicBool,
     runtime_handle: Option<tokio::runtime::Handle>,
     runtime_config: CoreRuntimeConfigStore,
@@ -107,12 +216,12 @@ where
     socket_context: SocketContext,
     events: Arc<dyn CoreEventSink>,
     tasks: Arc<std::sync::Mutex<JoinSet<()>>>,
-    cancel_tokens: Arc<DashMap<PortForwardConfig, DropGuard>>,
+    listeners: std::sync::Mutex<HashMap<(String, SocketAddr), PortForwardListener>>,
     udp_clients: Arc<DashMap<UdpClientKey, Arc<UdpClientInfo<H>>>>,
     udp_response_tasks: Arc<DashMap<UdpClientKey, AbortOnDropHandle<()>>>,
     udp_client_admission: Arc<Mutex<()>>,
     udp_client_slots: Arc<Semaphore>,
-    consumer_lease: Mutex<Option<DataPlaneConsumerLease>>,
+    consumer_lease: std::sync::Mutex<Option<DataPlaneConsumerLease>>,
 }
 
 impl<H> PortForwardAdapter<H>
@@ -128,7 +237,7 @@ where
         runtime_handle: Option<tokio::runtime::Handle>,
     ) -> Arc<Self> {
         Arc::new(Self {
-            operation: Mutex::new(()),
+            operation: Arc::new(Mutex::new(())),
             started: AtomicBool::new(false),
             runtime_handle,
             runtime_config,
@@ -137,17 +246,17 @@ where
             socket_context,
             events,
             tasks: Arc::new(std::sync::Mutex::new(JoinSet::new())),
-            cancel_tokens: Arc::new(DashMap::new()),
+            listeners: std::sync::Mutex::new(HashMap::new()),
             udp_clients: Arc::new(DashMap::new()),
             udp_response_tasks: Arc::new(DashMap::new()),
             udp_client_admission: Arc::new(Mutex::new(())),
             udp_client_slots: Arc::new(Semaphore::new(MAX_ACTIVE_UDP_CLIENTS)),
-            consumer_lease: Mutex::new(None),
+            consumer_lease: std::sync::Mutex::new(None),
         })
     }
 
-    pub(crate) async fn start(&self) -> anyhow::Result<()> {
-        let _operation = self.operation.lock().await;
+    pub(crate) async fn start(self: &Arc<Self>) -> anyhow::Result<()> {
+        let operation = self.operation.clone().lock_owned().await;
         if self.started.load(Ordering::Acquire) {
             return Ok(());
         }
@@ -163,110 +272,157 @@ where
             .gateway
             .port_forwards
             .clone();
-        if let Err(error) = self.apply_port_forwards(&cfgs).await {
-            self.stop_inner().await;
-            return Err(error);
-        }
+        let prepared = match self.prepare_inner(&cfgs, operation).await {
+            Ok(prepared) => prepared,
+            Err(error) => {
+                self.stop_inner().await;
+                return Err(error);
+            }
+        };
+        let cleanup = prepared.commit();
         self.started.store(true, Ordering::Release);
+        let _ = cleanup.await;
         Ok(())
     }
 
-    pub(crate) async fn reload(self: &Arc<Self>, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
+    pub(crate) async fn prepare(
+        self: &Arc<Self>,
+        cfgs: &[PortForwardConfig],
+    ) -> anyhow::Result<Option<PreparedPortForwards<H>>> {
         if let Some(runtime) = &self.runtime_handle {
             let adapter = self.clone();
             let cfgs = cfgs.to_vec();
-            // Binding and spawning must both happen on the instance executor,
-            // including reloads reached through management config patches.
+            // Binding must happen on the instance executor, including reloads
+            // reached through management config patches.
             return AbortOnDropHandle::new(
-                runtime.spawn(async move { adapter.reload_on_runtime(&cfgs).await }),
+                runtime.spawn(async move { adapter.prepare_on_runtime(&cfgs).await }),
             )
             .await
-            .map_err(|error| anyhow::anyhow!("port-forward reload task failed: {error}"))?;
+            .map_err(|error| anyhow::anyhow!("port-forward prepare task failed: {error}"))?;
         }
-        self.reload_on_runtime(cfgs).await
+        self.prepare_on_runtime(cfgs).await
     }
 
-    async fn reload_on_runtime(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
-        let _operation = self.operation.lock().await;
+    async fn prepare_on_runtime(
+        self: &Arc<Self>,
+        cfgs: &[PortForwardConfig],
+    ) -> anyhow::Result<Option<PreparedPortForwards<H>>> {
+        let operation = self.operation.clone().lock_owned().await;
         if !self.started.load(Ordering::Acquire) {
-            return Ok(());
+            return Ok(None);
         }
-        self.apply_port_forwards(cfgs).await
+        self.prepare_inner(cfgs, operation).await.map(Some)
     }
 
-    async fn apply_port_forwards(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
+    async fn prepare_inner(
+        self: &Arc<Self>,
+        cfgs: &[PortForwardConfig],
+        operation: OwnedMutexGuard<()>,
+    ) -> anyhow::Result<PreparedPortForwards<H>> {
+        let mut configs = HashMap::new();
         for cfg in cfgs {
-            if !matches!(cfg.proto.to_lowercase().as_str(), "tcp" | "udp") {
+            let mut cfg = cfg.clone();
+            cfg.proto.make_ascii_lowercase();
+            if !matches!(cfg.proto.as_str(), "tcp" | "udp") {
                 anyhow::bail!(
                     "unsupported protocol: {}, only support udp / tcp",
                     cfg.proto
                 );
             }
-        }
-
-        if !cfgs.is_empty() {
-            let mut consumer_lease = self.consumer_lease.lock().await;
-            if consumer_lease.is_none() {
-                consumer_lease.replace(self.data_plane.acquire_consumer_lease()?);
+            let key = (cfg.proto.clone(), cfg.bind_addr);
+            if let Some(previous) = configs.insert(key, cfg.clone()) {
+                anyhow::ensure!(
+                    previous == cfg,
+                    "conflicting port-forward targets for {} {}",
+                    cfg.proto,
+                    cfg.bind_addr
+                );
             }
         }
-
-        self.cancel_tokens.retain(|current, _| {
-            let keep = cfgs.iter().any(|next| {
-                if next.dst_addr.ip().is_unspecified() {
-                    current.bind_addr == next.bind_addr && current.proto == next.proto
-                } else {
-                    current == next
-                }
-            });
-            if !keep {
-                tracing::info!(?current, "port-forward removed by runtime config reload");
-            }
-            keep
-        });
-        self.udp_clients
-            .retain(|key, _| self.cancel_tokens.contains_key(&key.forward));
-        self.udp_response_tasks
-            .retain(|key, _| self.udp_clients.contains_key(key));
+        let consumer_lease = if !configs.is_empty() && self.consumer_lease.lock().unwrap().is_none()
+        {
+            Some(self.data_plane.acquire_consumer_lease()?)
+        } else {
+            None
+        };
+        let mut sockets = Vec::new();
         for cfg in cfgs {
-            if !self.cancel_tokens.contains_key(cfg) {
-                self.add_port_forward(cfg.clone()).await?;
+            let mut cfg = cfg.clone();
+            cfg.proto.make_ascii_lowercase();
+            let key = (cfg.proto.clone(), cfg.bind_addr);
+            if self.listeners.lock().unwrap().contains_key(&key)
+                || sockets
+                    .iter()
+                    .any(|(bound, _): &(PortForwardConfig, PortForwardSocket<H>)| {
+                        bound.proto == cfg.proto && bound.bind_addr == cfg.bind_addr
+                    })
+            {
+                continue;
             }
-        }
-        if cfgs.is_empty() {
-            self.consumer_lease.lock().await.take();
-        }
-        Ok(())
-    }
-
-    async fn add_port_forward(&self, cfg: PortForwardConfig) -> anyhow::Result<()> {
-        match cfg.proto.to_lowercase().as_str() {
-            "tcp" => self.add_tcp_port_forward(&cfg).await?,
-            "udp" => self.add_udp_port_forward(&cfg).await?,
-            _ => {
-                anyhow::bail!(
-                    "unsupported protocol: {}, only support udp / tcp",
-                    cfg.proto
+            let socket = if cfg.proto == "tcp" {
+                PortForwardSocket::Tcp(
+                    self.host
+                        .bind_tcp(self.tcp_listen_options(cfg.bind_addr))
+                        .await?,
                 )
-            }
+            } else {
+                PortForwardSocket::Udp(
+                    self.host
+                        .bind_udp(
+                            UdpBindOptions::port_forward(cfg.bind_addr)
+                                .with_context(self.socket_context.clone()),
+                        )
+                        .await?,
+                )
+            };
+            sockets.push((cfg, socket));
         }
-        self.events.emit(CoreEvent::GatewayPortForwardAdded(cfg));
-        Ok(())
+        Ok(PreparedPortForwards {
+            adapter: self.clone(),
+            operation,
+            configs,
+            sockets,
+            consumer_lease,
+        })
     }
 
-    async fn add_tcp_port_forward(&self, cfg: &PortForwardConfig) -> anyhow::Result<()> {
-        let (bind_addr, dst_addr) = (cfg.bind_addr, cfg.dst_addr);
+    fn tcp_listen_options(&self, bind_addr: SocketAddr) -> TcpListenOptions {
         let options = TcpListenOptions::port_forward(bind_addr);
         let bind = options
             .bind
             .clone()
             .with_context(self.socket_context.clone());
-        let options = options.with_bind(bind);
-        let listener = self.host.bind_tcp(options.clone()).await?;
+        options.with_bind(bind)
+    }
+
+    fn remove_udp_clients(&self, cfg: &PortForwardConfig, tasks: &mut Vec<AbortOnDropHandle<()>>) {
+        self.udp_clients.retain(|key, _| key.forward != *cfg);
+        let keys = self
+            .udp_response_tasks
+            .iter()
+            .filter(|entry| entry.key().forward == *cfg)
+            .map(|entry| entry.key().clone())
+            .collect::<Vec<_>>();
+        for key in keys {
+            if let Some((_, task)) = self.udp_response_tasks.remove(&key) {
+                task.abort();
+                tasks.push(task);
+            }
+        }
+    }
+
+    fn add_tcp_port_forward(
+        &self,
+        cfg: &PortForwardConfig,
+        listener: Arc<H::Listener>,
+    ) -> PortForwardListener {
+        let (bind_addr, dst_addr) = (cfg.bind_addr, cfg.dst_addr);
+        let options = self.tcp_listen_options(bind_addr);
         tracing::info!(?bind_addr, ?dst_addr, "TCP port-forward listener bound");
         let cancel = CancellationToken::new();
-        self.cancel_tokens
-            .insert(cfg.clone(), cancel.clone().drop_guard());
+        let config = Arc::new(std::sync::Mutex::new(cfg.clone()));
+        let task_cancel = cancel.clone();
+        let task_config = config.clone();
 
         let data_plane = self.data_plane.clone();
         let connections = Arc::new(std::sync::Mutex::new(JoinSet::new()));
@@ -275,7 +431,8 @@ where
             "TCP port-forward connections",
         ));
         let host = self.host.clone();
-        self.tasks.lock().unwrap().spawn(async move {
+        let task = AbortOnDropHandle::new(tokio::spawn(async move {
+            let cancel = task_cancel;
             let mut listener = Some(listener);
             loop {
                 let Some((incoming, source_addr)) =
@@ -283,6 +440,7 @@ where
                 else {
                     break;
                 };
+                let dst_addr = task_config.lock().unwrap().dst_addr;
                 tracing::info!(
                     ?bind_addr,
                     ?source_addr,
@@ -317,22 +475,27 @@ where
                 ?dst_addr,
                 "TCP port-forward listener task stopped"
             );
-        });
-        Ok(())
+        }));
+        PortForwardListener {
+            config,
+            cancel,
+            task,
+            udp_tasks: TaskTracker::new(),
+        }
     }
 
-    async fn add_udp_port_forward(&self, cfg: &PortForwardConfig) -> anyhow::Result<()> {
-        let (bind_addr, dst_addr) = (cfg.bind_addr, cfg.dst_addr);
-        let forward = cfg.clone();
-        let socket = self
-            .host
-            .bind_udp(
-                UdpBindOptions::port_forward(bind_addr).with_context(self.socket_context.clone()),
-            )
-            .await?;
+    fn add_udp_port_forward(
+        &self,
+        cfg: &PortForwardConfig,
+        socket: Arc<<H as VirtualUdpSocketFactory>::Socket>,
+    ) -> PortForwardListener {
+        let bind_addr = cfg.bind_addr;
         let cancel = CancellationToken::new();
-        self.cancel_tokens
-            .insert(cfg.clone(), cancel.clone().drop_guard());
+        let config = Arc::new(std::sync::Mutex::new(cfg.clone()));
+        let task_cancel = cancel.clone();
+        let task_config = config.clone();
+        let udp_tasks = TaskTracker::new();
+        let response_tracker = udp_tasks.clone();
 
         let data_plane = self.data_plane.clone();
         let host = self.host.clone();
@@ -341,7 +504,8 @@ where
         let response_tasks = self.udp_response_tasks.clone();
         let client_admission = self.udp_client_admission.clone();
         let client_slots = self.udp_client_slots.clone();
-        self.tasks.lock().unwrap().spawn(async move {
+        let task = AbortOnDropHandle::new(tokio::spawn(async move {
+            let cancel = task_cancel;
             let adapter = UdpFlowFactory {
                 data_plane,
                 host,
@@ -360,6 +524,8 @@ where
                         }
                     },
                 };
+                let forward = task_config.lock().unwrap().clone();
+                let dst_addr = forward.dst_addr;
                 let key = UdpClientKey {
                     client_addr,
                     forward: forward.clone(),
@@ -394,6 +560,13 @@ where
                             continue;
                         };
                         let UdpClientReservation { slot, admission } = reservation;
+                        // Commit uses this same lock when replacing the target
+                        // and removing its flows. Do not publish an old flow
+                        // after either operation, including after an async open.
+                        let current = task_config.lock().unwrap();
+                        if *current != forward || cancel.is_cancelled() {
+                            continue;
+                        }
                         let slot = Arc::new(slot);
                         let client = Arc::new(UdpClientInfo {
                             flow: flow.clone(),
@@ -408,7 +581,7 @@ where
                         let response_slot = slot;
                         response_tasks.insert(
                             key.clone(),
-                            AbortOnDropHandle::new(tokio::spawn(async move {
+                            AbortOnDropHandle::new(response_tracker.spawn(async move {
                                 let _slot = response_slot;
                                 let mut buf = vec![0u8; MAX_UDP_PAYLOAD_SIZE];
                                 loop {
@@ -435,6 +608,7 @@ where
                                 }
                             })),
                         );
+                        drop(current);
                         drop(admission);
                         client
                     }
@@ -444,9 +618,14 @@ where
                     tracing::error!(?error, ?dst_addr, "send UDP data-plane packet failed");
                 }
             }
-        });
+        }));
 
-        Ok(())
+        PortForwardListener {
+            config,
+            cancel,
+            task,
+            udp_tasks,
+        }
     }
 
     fn start_udp_reaper(&self) {
@@ -468,14 +647,29 @@ where
 
     async fn stop_inner(&self) {
         self.started.store(false, Ordering::Release);
+        let listeners = std::mem::take(&mut *self.listeners.lock().unwrap());
         tracing::info!(
-            forward_count = self.cancel_tokens.len(),
+            forward_count = listeners.len(),
             "port-forward adapter stopping"
         );
-        self.cancel_tokens.clear();
+        for listener in listeners.values() {
+            listener.cancel.cancel();
+            listener.task.abort();
+        }
+        // First stop publishers, then remove response handles. The trackers
+        // include responses whose handles the reaper/eviction already dropped.
+        let mut trackers = Vec::new();
+        for (_, listener) in listeners {
+            let _ = listener.task.await;
+            trackers.push(listener.udp_tasks);
+        }
         self.udp_response_tasks.clear();
         self.udp_clients.clear();
-        self.consumer_lease.lock().await.take();
+        for tracker in trackers {
+            tracker.close();
+            tracker.wait().await;
+        }
+        self.consumer_lease.lock().unwrap().take();
         let mut tasks = {
             let mut tasks = self.tasks.lock().unwrap();
             std::mem::replace(&mut *tasks, JoinSet::new())
@@ -639,6 +833,9 @@ where
     }
 }
 
+#[cfg(test)]
+mod reload_tests;
+
 #[cfg(test)]
 mod tests {
     use std::{io, sync::atomic::AtomicUsize};

easytier-core/src/gateway/port_forward/reload_tests.rs

@@ -0,0 +1,290 @@
+use std::{io, sync::Weak};
+
+use tokio::sync::{Notify, mpsc, oneshot};
+
+use super::*;
+use crate::{
+    config::{IpPrefix, peers::PeerRuntimeSnapshot, runtime::CoreRuntimeConfig},
+    host::{
+        packet::{HostPacketReceiver, host_packet_channel},
+        testkit::TestHost,
+    },
+    peers::peer_manager::{PeerManagerCore, PortablePeerManagerConfig},
+    socket::tcp::TcpConnectOptions,
+};
+
+type Datagram = (Vec<u8>, SocketAddr);
+
+struct ReloadUdpSocket {
+    address: SocketAddr,
+    incoming: Option<Mutex<mpsc::UnboundedReceiver<Datagram>>>,
+    sent: mpsc::UnboundedSender<Datagram>,
+}
+
+#[async_trait::async_trait]
+impl VirtualUdpSocket for ReloadUdpSocket {
+    fn local_addr(&self) -> io::Result<SocketAddr> {
+        Ok(self.address)
+    }
+
+    async fn send_to(&self, data: &[u8], address: SocketAddr) -> io::Result<usize> {
+        self.sent.send((data.to_vec(), address)).unwrap();
+        Ok(data.len())
+    }
+
+    async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
+        let Some(incoming) = &self.incoming else {
+            return std::future::pending().await;
+        };
+        let (data, address) = incoming.lock().await.recv().await.unwrap();
+        buf[..data.len()].copy_from_slice(&data);
+        Ok((data.len(), address))
+    }
+}
+
+struct ReloadUdpHost {
+    incoming: std::sync::Mutex<Option<mpsc::UnboundedReceiver<Datagram>>>,
+    sent: mpsc::UnboundedSender<Datagram>,
+    sockets: std::sync::Mutex<Vec<Weak<ReloadUdpSocket>>>,
+    open_started: Notify,
+    open_release: std::sync::Mutex<Option<oneshot::Receiver<()>>>,
+}
+
+#[async_trait::async_trait]
+impl VirtualTcpSocketFactory for ReloadUdpHost {
+    type Socket = <TestHost as VirtualTcpSocketFactory>::Socket;
+
+    async fn connect_tcp(&self, options: TcpConnectOptions) -> anyhow::Result<Self::Socket> {
+        TestHost::default().connect_tcp(options).await
+    }
+}
+
+#[async_trait::async_trait]
+impl VirtualTcpListenerFactory for ReloadUdpHost {
+    type Listener = <TestHost as VirtualTcpListenerFactory>::Listener;
+
+    async fn bind_tcp(&self, options: TcpListenOptions) -> anyhow::Result<Arc<Self::Listener>> {
+        TestHost::default().bind_tcp(options).await
+    }
+}
+
+#[async_trait::async_trait]
+impl VirtualUdpSocketFactory for ReloadUdpHost {
+    type Socket = ReloadUdpSocket;
+
+    async fn bind_udp(&self, options: UdpBindOptions) -> anyhow::Result<Arc<Self::Socket>> {
+        let address = options.local_addr.unwrap();
+        let incoming = if address.port() != 0 {
+            self.incoming.lock().unwrap().take().map(Mutex::new)
+        } else {
+            let release = self.open_release.lock().unwrap().take();
+            if let Some(release) = release {
+                self.open_started.notify_one();
+                release.await.unwrap();
+            }
+            None
+        };
+        let socket = Arc::new(ReloadUdpSocket {
+            address,
+            incoming,
+            sent: self.sent.clone(),
+        });
+        self.sockets.lock().unwrap().push(Arc::downgrade(&socket));
+        Ok(socket)
+    }
+}
+
+struct UdpReloadTest {
+    adapter: Arc<PortForwardAdapter<ReloadUdpHost>>,
+    host: Arc<ReloadUdpHost>,
+    incoming: mpsc::UnboundedSender<Datagram>,
+    sent: mpsc::UnboundedReceiver<Datagram>,
+    _peer_manager: Arc<PeerManagerCore>,
+    _packets: HostPacketReceiver,
+}
+
+impl UdpReloadTest {
+    async fn new() -> Self {
+        let mut runtime = PeerRuntimeSnapshot::default().runtime;
+        runtime.core.routes.ipv4 = Some(IpPrefix::new("10.144.0.2".parse().unwrap(), 24).unwrap());
+        let peer_config = PortablePeerManagerConfig::new(runtime);
+        let runtime_config = CoreRuntimeConfigStore::new(
+            CoreRuntimeConfig::default(),
+            Arc::new(peer_config.snapshot.clone()),
+        );
+        let (packet_sender, packets) = host_packet_channel();
+        let peer_manager =
+            Arc::new(PeerManagerCore::new_portable_for_test(peer_config, packet_sender).unwrap());
+        let (incoming, receiver) = mpsc::unbounded_channel();
+        let (sent, sent_receiver) = mpsc::unbounded_channel();
+        let host = Arc::new(ReloadUdpHost {
+            incoming: std::sync::Mutex::new(Some(receiver)),
+            sent,
+            sockets: std::sync::Mutex::new(Vec::new()),
+            open_started: Notify::new(),
+            open_release: std::sync::Mutex::new(None),
+        });
+        let data_plane = DataPlaneRuntime::new(
+            runtime_config.clone(),
+            peer_manager.clone(),
+            None,
+            host.clone(),
+            SocketContext::default(),
+        );
+        data_plane.start_runtime().await.unwrap();
+        let adapter = PortForwardAdapter::new(
+            runtime_config,
+            data_plane,
+            host.clone(),
+            SocketContext::default(),
+            Arc::new(()),
+            None,
+        );
+        adapter.start().await.unwrap();
+        Self {
+            adapter,
+            host,
+            incoming,
+            sent: sent_receiver,
+            _peer_manager: peer_manager,
+            _packets: packets,
+        }
+    }
+
+    async fn apply(&self, port: u16) -> PortForwardConfig {
+        let config = PortForwardConfig {
+            proto: "udp".to_owned(),
+            bind_addr: "127.0.0.1:19491".parse().unwrap(),
+            dst_addr: SocketAddr::from(([10, 144, 0, 2], port)),
+        };
+        self.adapter
+            .prepare(std::slice::from_ref(&config))
+            .await
+            .unwrap()
+            .unwrap()
+            .commit()
+            .await
+            .unwrap();
+        config
+    }
+
+    fn send(&self, payload: &[u8]) {
+        self.incoming
+            .send((payload.to_vec(), "127.0.0.1:40000".parse().unwrap()))
+            .unwrap();
+    }
+
+    async fn receive(&mut self) -> Datagram {
+        tokio::time::timeout(Duration::from_secs(2), self.sent.recv())
+            .await
+            .unwrap()
+            .unwrap()
+    }
+
+    async fn stop(&self) {
+        self.adapter.stop().await;
+        self.adapter.data_plane.stop_runtime().await;
+    }
+}
+
+#[tokio::test]
+async fn udp_target_update_reuses_listener_and_releases_old_flow() {
+    let mut test = UdpReloadTest::new().await;
+    let first = test.apply(19381).await;
+    test.send(b"first");
+    assert_eq!(test.receive().await, (b"first".to_vec(), first.dst_addr));
+    let second = test.apply(19382).await;
+    assert!(test.adapter.udp_clients.is_empty());
+    {
+        let sockets = test.host.sockets.lock().unwrap();
+        assert_eq!(
+            sockets.len(),
+            2,
+            "target changes must not bind a new listener"
+        );
+        assert!(sockets[0].upgrade().is_some());
+        assert!(sockets[1].upgrade().is_none(), "old flow must be released");
+    }
+    test.send(b"second");
+    assert_eq!(test.receive().await, (b"second".to_vec(), second.dst_addr));
+    assert_eq!(test.adapter.udp_clients.len(), 1);
+    test.stop().await;
+}
+
+#[tokio::test]
+async fn udp_target_update_during_open_discards_old_flow() {
+    let mut test = UdpReloadTest::new().await;
+    test.apply(19381).await;
+    let (release, waiting) = oneshot::channel();
+    *test.host.open_release.lock().unwrap() = Some(waiting);
+    test.send(b"old");
+    tokio::time::timeout(Duration::from_secs(2), test.host.open_started.notified())
+        .await
+        .unwrap();
+    let next = test.apply(19382).await;
+    release.send(()).unwrap();
+    test.send(b"new");
+    assert_eq!(test.receive().await, (b"new".to_vec(), next.dst_addr));
+    assert!(
+        test.adapter
+            .udp_clients
+            .iter()
+            .all(|entry| entry.key().forward == next)
+    );
+    assert!(test.host.sockets.lock().unwrap()[1].upgrade().is_none());
+    test.stop().await;
+}
+
+#[cfg(not(any(target_arch = "wasm32", target_os = "wasi")))]
+#[tokio::test]
+async fn udp_removal_waits_for_response_tasks_already_removed_from_map() {
+    let test = UdpReloadTest::new().await;
+    let forward = test.apply(19381).await;
+    let socket = test.host.sockets.lock().unwrap()[0].upgrade().unwrap();
+    let tracker = test
+        .adapter
+        .listeners
+        .lock()
+        .unwrap()
+        .get(&(forward.proto.clone(), forward.bind_addr))
+        .unwrap()
+        .udp_tasks
+        .clone();
+    let (started, ready) = oneshot::channel();
+    let (release, waiting) = std::sync::mpsc::channel();
+    // A running blocking task makes the delay between abort and socket release
+    // deterministic, as can happen when a response task is currently executing.
+    let task = tracker.spawn_blocking(move || {
+        started.send(()).unwrap();
+        waiting.recv().unwrap();
+        drop(socket);
+    });
+    let key = UdpClientKey {
+        client_addr: "127.0.0.1:40000".parse().unwrap(),
+        forward,
+    };
+    test.adapter
+        .udp_response_tasks
+        .insert(key, AbortOnDropHandle::new(task));
+    ready.await.unwrap();
+    test.adapter.udp_response_tasks.clear();
+    let cleanup = test.adapter.prepare(&[]).await.unwrap().unwrap().commit();
+    drop(cleanup);
+    let mut next = std::pin::pin!(test.adapter.prepare(&[]));
+    let result = tokio::time::timeout(Duration::from_millis(20), &mut next).await;
+    let pending = result.is_err();
+    release.send(()).unwrap();
+    let prepared = match result {
+        Ok(prepared) => prepared,
+        Err(_) => tokio::time::timeout(Duration::from_secs(2), next)
+            .await
+            .unwrap(),
+    };
+    prepared.unwrap().unwrap().commit().await.unwrap();
+    assert!(
+        pending,
+        "dropping the cleanup waiter released the operation lock before the socket"
+    );
+    assert!(test.host.sockets.lock().unwrap()[0].upgrade().is_none());
+    test.stop().await;
+}

easytier-core/src/host/testkit.rs

@@ -6,11 +6,12 @@
 //! accept, UDP sockets never receive, and DNS answers only literal IP hosts.
 
 use std::{
+    collections::HashMap,
     io,
     net::{IpAddr, SocketAddr},
     pin::Pin,
     sync::{
-        Arc,
+        Arc, Mutex,
         atomic::{AtomicUsize, Ordering},
     },
     task::{Context, Poll},
@@ -70,11 +71,16 @@ impl VirtualTcpSocket for TestTcpSocket {
 pub struct TestTcpListener {
     address: SocketAddr,
     active_listeners: Arc<AtomicUsize>,
+    listeners: Option<Arc<Mutex<TestTcpListeners>>>,
+    incoming: Option<tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<TestTcpSocket>>>,
 }
 
 impl Drop for TestTcpListener {
     fn drop(&mut self) {
         self.active_listeners.fetch_sub(1, Ordering::Relaxed);
+        if let Some(listeners) = &self.listeners {
+            listeners.lock().unwrap().remove(&self.address);
+        }
     }
 }
 
@@ -87,6 +93,12 @@ impl VirtualTcpListener for TestTcpListener {
     }
 
     async fn accept(&self) -> io::Result<(Self::Socket, SocketAddr)> {
+        if let Some(incoming) = &self.incoming {
+            let socket = incoming.lock().await.recv().await.ok_or_else(|| {
+                io::Error::new(io::ErrorKind::ConnectionAborted, "test listener closed")
+            })?;
+            return Ok((socket, "127.0.0.1:20001".parse().unwrap()));
+        }
         std::future::pending().await
     }
 }
@@ -108,6 +120,8 @@ impl VirtualUdpSocket for TestUdpSocket {
     }
 }
 
+pub type TestTcpListeners = HashMap<SocketAddr, tokio::sync::mpsc::UnboundedSender<TestTcpSocket>>;
+
 #[derive(Default)]
 pub struct TestHost {
     pub tcp_binds: AtomicUsize,
@@ -115,21 +129,42 @@ pub struct TestHost {
     pub udp_binds: AtomicUsize,
     pub proxy_nat_connections:
         Option<tokio::sync::mpsc::UnboundedSender<(SocketAddr, tokio::io::DuplexStream)>>,
+    pub port_forward_connections:
+        Option<tokio::sync::mpsc::UnboundedSender<(SocketAddr, tokio::io::DuplexStream)>>,
+    pub tcp_listeners: Option<Arc<Mutex<TestTcpListeners>>>,
+    pub reject_tcp_bind: Mutex<Option<SocketAddr>>,
     pub reject_socks5_listener: bool,
 }
 
+impl TestHost {
+    #[cfg(all(feature = "management", feature = "proxy-smoltcp-stack"))]
+    pub fn connect_to_listener(&self, address: SocketAddr) -> tokio::io::DuplexStream {
+        let (socket, peer) = tokio::io::duplex(1024);
+        self.tcp_listeners
+            .as_ref()
+            .unwrap()
+            .lock()
+            .unwrap()
+            .get(&address)
+            .expect("test listener must be bound")
+            .send(TestTcpSocket(socket))
+            .unwrap();
+        peer
+    }
+}
+
 #[async_trait::async_trait]
 impl VirtualTcpSocketFactory for TestHost {
     type Socket = TestTcpSocket;
 
     async fn connect_tcp(&self, options: TcpConnectOptions) -> anyhow::Result<Self::Socket> {
-        if options.purpose != TcpSocketPurpose::ProxyNat {
-            anyhow::bail!("test host does not connect non-proxy TCP sockets");
+        let connections = match options.purpose {
+            TcpSocketPurpose::ProxyNat => &self.proxy_nat_connections,
+            TcpSocketPurpose::PortForward => &self.port_forward_connections,
+            _ => anyhow::bail!("test host does not connect non-proxy TCP sockets"),
         }
-        let connections = self
-            .proxy_nat_connections
-            .as_ref()
-            .ok_or_else(|| anyhow::anyhow!("test host proxy NAT is disabled"))?;
+        .as_ref()
+        .ok_or_else(|| anyhow::anyhow!("test host TCP connections are disabled"))?;
         let (socket, peer) = tokio::io::duplex(1024);
         connections
             .send((options.remote_addr, peer))
@@ -159,10 +194,26 @@ impl VirtualTcpListenerFactory for TestHost {
         } else {
             address
         };
+        if *self.reject_tcp_bind.lock().unwrap() == Some(address) {
+            anyhow::bail!("test host rejected TCP bind to {address}");
+        }
+        let incoming = if let Some(listeners) = &self.tcp_listeners {
+            let mut listeners = listeners.lock().unwrap();
+            if listeners.contains_key(&address) {
+                return Err(io::Error::new(io::ErrorKind::AddrInUse, address.to_string()).into());
+            }
+            let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
+            listeners.insert(address, sender);
+            Some(tokio::sync::Mutex::new(receiver))
+        } else {
+            None
+        };
         self.active_tcp_listeners.fetch_add(1, Ordering::Relaxed);
         Ok(Arc::new(TestTcpListener {
             address,
             active_listeners: self.active_tcp_listeners.clone(),
+            listeners: self.tcp_listeners.clone(),
+            incoming,
         }))
     }
 }

easytier-core/src/instance/manager.rs

@@ -387,7 +387,11 @@ where
         let instance_state_changes = self.instance_state_changes.clone();
         runtime.spawn(async move {
             if let Err(error) = instance.start().await {
-                tracing::error!(%error, %instance_id, "instance failed to start");
+                tracing::error!(
+                    error = %format_args!("{error:#}"),
+                    %instance_id,
+                    "instance failed to start"
+                );
             }
             instance_state_changes.mark_changed();
             notifier.notify_one();

easytier-core/src/instance/mod.rs

@@ -936,26 +936,57 @@ where
             anyhow::bail!("runtime config cannot update while instance is stopping or stopped");
         }
         self.validate_runtime_config_capabilities(&config)?;
+        #[cfg(feature = "proxy-smoltcp-stack")]
+        let prepared = self.prepare_port_forwards(&config).await?;
+        self.commit_runtime_config_under_operation(
+            config,
+            #[cfg(feature = "proxy-smoltcp-stack")]
+            prepared,
+            || (),
+        )
+        .await
+    }
+
+    #[cfg(feature = "proxy-smoltcp-stack")]
+    pub(crate) async fn prepare_port_forwards(
+        &self,
+        config: &CoreInstanceRuntimeConfig,
+    ) -> anyhow::Result<Option<crate::gateway::PreparedPortForwards<H>>> {
+        self.port_forward_adapter
+            .prepare(&config.services.gateway.port_forwards)
+            .await
+    }
+
+    pub(crate) async fn commit_runtime_config_under_operation(
+        &self,
+        config: CoreInstanceRuntimeConfig,
+        #[cfg(feature = "proxy-smoltcp-stack")] prepared: Option<
+            crate::gateway::PreparedPortForwards<H>,
+        >,
+        commit_config: impl FnOnce(),
+    ) -> anyhow::Result<()> {
         #[cfg(feature = "test-utils")]
         let reload_acl = self.runtime_config.snapshot().services.acl != config.services.acl;
-        let published = self.peer_manager.update_runtime_config(config).await?;
+        let (published, cleanup) = self
+            .peer_manager
+            .update_runtime_config_with(config, || {
+                #[cfg(feature = "proxy-smoltcp-stack")]
+                let cleanup = prepared.map(|prepared| prepared.commit());
+                #[cfg(not(feature = "proxy-smoltcp-stack"))]
+                let cleanup: Option<tokio::task::JoinHandle<()>> = None;
+                commit_config();
+                cleanup
+            })
+            .await?;
         #[cfg(feature = "test-utils")]
         if reload_acl {
             self.acl_reload_count.fetch_add(1, Ordering::Relaxed);
         }
         self.proxy_cidr_table
             .update_snapshot(proxy_cidr_snapshot(&published));
-        #[cfg(feature = "proxy-smoltcp-stack")]
-        self.port_forward_adapter
-            .reload(
-                &self
-                    .runtime_config
-                    .snapshot()
-                    .services
-                    .gateway
-                    .port_forwards,
-            )
-            .await?;
+        if let Some(cleanup) = cleanup {
+            let _ = cleanup.await;
+        }
         Ok(())
     }
 

easytier-core/src/instance/tests.rs

@@ -2981,6 +2981,268 @@ virtual_ip = "10.82.0.2/24"
         assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
     }
 
+    #[cfg(all(feature = "management", feature = "proxy-smoltcp-stack"))]
+    mod port_forward_reload {
+        use super::*;
+        use easytier_proto::api::config::{
+            ConfigPatchAction, InstanceConfigPatch, PortForwardPatch,
+        };
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        use tokio::sync::mpsc::UnboundedReceiver;
+
+        fn forward(bind_port: u16, destination_port: u16) -> PortForwardConfig {
+            PortForwardConfig {
+                bind_addr: SocketAddr::from(([127, 0, 0, 1], bind_port)),
+                dst_addr: SocketAddr::from(([10, 144, 0, 2], destination_port)),
+                proto: "tcp".to_owned(),
+            }
+        }
+
+        async fn start() -> (
+            Arc<CoreInstance<TestHost>>,
+            Arc<TestHost>,
+            UnboundedReceiver<(SocketAddr, tokio::io::DuplexStream)>,
+        ) {
+            let (connections, receiver) = tokio::sync::mpsc::unbounded_channel();
+            let host = Arc::new(TestHost {
+                port_forward_connections: Some(connections),
+                tcp_listeners: Some(Arc::new(std::sync::Mutex::new(Default::default()))),
+                ..Default::default()
+            });
+            let config =
+                TomlConfig::new_from_str("instance_name = \"port-forward-reload\"").unwrap();
+            config.set_port_forwards(vec![forward(19491, 19381)]);
+            let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
+            let instance = CoreInstance::from_toml(
+                config,
+                adapters_with_host(host.clone(), None, Arc::new(packet_sink)),
+            )
+            .unwrap();
+            instance.start().await.unwrap();
+            (instance, host, receiver)
+        }
+
+        async fn assert_destination(
+            host: &TestHost,
+            receiver: &mut UnboundedReceiver<(SocketAddr, tokio::io::DuplexStream)>,
+            config: &PortForwardConfig,
+        ) -> (tokio::io::DuplexStream, tokio::io::DuplexStream) {
+            let mut incoming = host.connect_to_listener(config.bind_addr);
+            let outgoing = tokio::time::timeout(Duration::from_secs(2), async {
+                let (destination, mut outgoing) = receiver.recv().await.unwrap();
+                assert_eq!(destination, config.dst_addr);
+                incoming.write_all(b"request").await.unwrap();
+                let mut data = [0; 7];
+                outgoing.read_exact(&mut data).await.unwrap();
+                assert_eq!(&data, b"request");
+                outgoing
+            })
+            .await
+            .expect("port forward must accept and forward to its committed destination");
+            (incoming, outgoing)
+        }
+
+        fn patch(
+            removed: &[PortForwardConfig],
+            added: &[PortForwardConfig],
+        ) -> InstanceConfigPatch {
+            InstanceConfigPatch {
+                port_forwards: removed
+                    .iter()
+                    .map(|config| (ConfigPatchAction::Remove, config))
+                    .chain(added.iter().map(|config| (ConfigPatchAction::Add, config)))
+                    .map(|(action, config)| PortForwardPatch {
+                        action: action as i32,
+                        cfg: Some(config.clone().into()),
+                    })
+                    .collect(),
+                ..Default::default()
+            }
+        }
+
+        fn persistence(fail: bool) -> RecordingConfigPatchPersistence {
+            RecordingConfigPatchPersistence {
+                writes: std::sync::Mutex::new(Vec::new()),
+                fail: AtomicBool::new(fail),
+            }
+        }
+
+        #[tokio::test]
+        async fn destination_changes_reuse_listener() {
+            let (instance, host, mut connections) = start().await;
+            let (mut established_client, mut established_destination) =
+                assert_destination(&host, &mut connections, &forward(19491, 19381)).await;
+            let initial_binds = host.tcp_binds.load(Ordering::Relaxed);
+
+            for port in [19382, 19383, 19381] {
+                let config = forward(19491, port);
+                let mut runtime = (*instance.runtime_config.snapshot()).clone();
+                runtime.services.gateway.port_forwards = vec![config.clone()];
+                instance.update_runtime_config(runtime).await.unwrap();
+
+                assert_eq!(host.tcp_binds.load(Ordering::Relaxed), initial_binds);
+                assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+                assert_destination(&host, &mut connections, &config).await;
+                tokio::time::timeout(Duration::from_secs(2), async {
+                    let payload = port.to_be_bytes();
+                    established_client.write_all(&payload).await.unwrap();
+                    let mut received = [0; 2];
+                    established_destination
+                        .read_exact(&mut received)
+                        .await
+                        .unwrap();
+                    assert_eq!(received, payload);
+                    established_destination.write_all(&payload).await.unwrap();
+                    established_client.read_exact(&mut received).await.unwrap();
+                    assert_eq!(received, payload);
+                })
+                .await
+                .expect("changing the destination must preserve established TCP connections");
+            }
+            instance.stop().await;
+        }
+
+        #[tokio::test]
+        async fn failed_runtime_bind_keeps_committed_snapshot() {
+            let (instance, host, mut connections) = start().await;
+            let original = forward(19491, 19381);
+            let rejected = forward(19492, 19382);
+            *host.reject_tcp_bind.lock().unwrap() = Some(rejected.bind_addr);
+            let mut runtime = (*instance.runtime_config.snapshot()).clone();
+            runtime.services.gateway.port_forwards = vec![original.clone(), rejected];
+
+            assert!(
+                instance
+                    .update_runtime_config(runtime.clone())
+                    .await
+                    .is_err()
+            );
+            assert_eq!(
+                instance
+                    .runtime_config
+                    .snapshot()
+                    .services
+                    .gateway
+                    .port_forwards,
+                std::slice::from_ref(&original)
+            );
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            assert_destination(&host, &mut connections, &original).await;
+
+            *host.reject_tcp_bind.lock().unwrap() = None;
+            instance.update_runtime_config(runtime).await.unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 2);
+            instance.stop().await;
+        }
+
+        #[tokio::test]
+        async fn failed_patch_bind_releases_prepared_listener_and_retries() {
+            let (instance, host, mut connections) = start().await;
+            let original = forward(19491, 19381);
+            let added = [forward(19492, 19382), forward(19493, 19383)];
+            let before = instance.toml_config().unwrap().dump();
+            let persistence = persistence(false);
+            let patch = patch(&[], &added);
+            *host.reject_tcp_bind.lock().unwrap() = Some(added[1].bind_addr);
+            let initial_binds = host.tcp_binds.load(Ordering::Relaxed);
+
+            let error =
+                crate::management::apply_config_patch(&instance, patch.clone(), Some(&persistence))
+                    .await
+                    .unwrap_err();
+
+            assert!(error.to_string().contains("test host rejected TCP bind"));
+            assert_eq!(host.tcp_binds.load(Ordering::Relaxed), initial_binds + 2);
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            assert_eq!(instance.toml_config().unwrap().dump(), before);
+            assert_eq!(
+                instance
+                    .runtime_config
+                    .snapshot()
+                    .services
+                    .gateway
+                    .port_forwards,
+                std::slice::from_ref(&original)
+            );
+            assert!(persistence.writes.lock().unwrap().is_empty());
+            assert_destination(&host, &mut connections, &original).await;
+
+            *host.reject_tcp_bind.lock().unwrap() = None;
+            crate::management::apply_config_patch(&instance, patch, Some(&persistence))
+                .await
+                .unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 3);
+            assert_eq!(persistence.writes.lock().unwrap().len(), 1);
+            for config in &added {
+                assert_destination(&host, &mut connections, config).await;
+            }
+            instance.stop().await;
+        }
+
+        #[tokio::test]
+        async fn failed_patch_persistence_keeps_destination_and_releases_prepared_listener() {
+            let (instance, host, mut connections) = start().await;
+            let original = forward(19491, 19381);
+            let replacement = forward(19491, 19382);
+            let added = forward(19492, 19383);
+            let before = instance.toml_config().unwrap().dump();
+            let persistence = persistence(true);
+            let patch = patch(
+                std::slice::from_ref(&original),
+                &[replacement.clone(), added.clone()],
+            );
+
+            let error =
+                crate::management::apply_config_patch(&instance, patch.clone(), Some(&persistence))
+                    .await
+                    .unwrap_err();
+
+            assert!(
+                error
+                    .to_string()
+                    .contains("injected config persistence failure")
+            );
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            assert_eq!(instance.toml_config().unwrap().dump(), before);
+            assert_eq!(
+                instance
+                    .runtime_config
+                    .snapshot()
+                    .services
+                    .gateway
+                    .port_forwards,
+                std::slice::from_ref(&original)
+            );
+            assert!(persistence.writes.lock().unwrap().is_empty());
+            assert_destination(&host, &mut connections, &original).await;
+
+            persistence.fail.store(false, Ordering::Relaxed);
+            crate::management::apply_config_patch(&instance, patch, Some(&persistence))
+                .await
+                .unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 2);
+            assert_eq!(persistence.writes.lock().unwrap().len(), 1);
+            assert_destination(&host, &mut connections, &replacement).await;
+            assert_destination(&host, &mut connections, &added).await;
+            instance.stop().await;
+        }
+
+        #[tokio::test]
+        async fn removed_listener_can_be_rebound_immediately() {
+            let (instance, host, mut connections) = start().await;
+            let original = (*instance.runtime_config.snapshot()).clone();
+            let mut empty = original.clone();
+            empty.services.gateway.port_forwards.clear();
+
+            instance.update_runtime_config(empty).await.unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
+            instance.update_runtime_config(original).await.unwrap();
+            assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1);
+            assert_destination(&host, &mut connections, &forward(19491, 19381)).await;
+            instance.stop().await;
+        }
+    }
+
     #[cfg(feature = "proxy-packet")]
     #[tokio::test]
     async fn runtime_core_instance_owns_wrapped_transport_source_nat() {

easytier-core/src/listener/mod.rs

@@ -342,10 +342,11 @@ where
     Accepted: Send + 'static,
 {
     let mut listener = creator();
-    match listener.listen().await {
-        Ok(()) => Ok(listener),
-        Err(error) => Err(error),
-    }
+    listener
+        .listen()
+        .await
+        .with_context(|| format!("failed to listen on {}", listener.local_url()))?;
+    Ok(listener)
 }
 
 async fn run_listener<Accepted, H>(
@@ -782,7 +783,8 @@ mod tests {
             true,
         );
 
-        assert!(manager.run().await.is_err());
+        let error = manager.run().await.unwrap_err();
+        assert!(format!("{error:#}").contains("mock://second"));
         crate::foundation::time::sleep(Duration::from_millis(20)).await;
         assert!(handler.accepted.lock().unwrap().is_empty());
         assert!(

easytier-core/src/management/full/config_patch.rs

@@ -51,9 +51,25 @@ where
     // Preserve the existing ordered partial-commit contract: earlier valid
     // sub-patches remain applied if a later sub-patch fails.
     let patch_result: anyhow::Result<(bool, bool)> = async {
-        let result = patch_port_forwards(&candidate, patch.port_forwards);
-        validate_persist_and_commit_candidate(instance, &config, &candidate, persistence).await?;
-        result?;
+        if !patch.port_forwards.is_empty() {
+            patch_port_forwards(&candidate, patch.port_forwards)?;
+            // Port forwards have fallible host resources. Reserve them before the
+            // durable write, then commit resources and readback together. A failed
+            // bind or persistence operation must leave the previous rule usable.
+            let normalized = validate_candidate(instance, &candidate)?;
+            let runtime = runtime_config_from_normalized(&normalized);
+            #[cfg(feature = "proxy-smoltcp-stack")]
+            let prepared = instance.prepare_port_forwards(&runtime).await?;
+            persist_candidate_if_changed(instance, &config, &candidate, persistence).await?;
+            instance
+                .commit_runtime_config_under_operation(
+                    runtime,
+                    #[cfg(feature = "proxy-smoltcp-stack")]
+                    prepared,
+                    || config.replace_from_snapshot(&candidate),
+                )
+                .await?;
+        }
 
         let result = patch_acl(&candidate, patch.acl);
         validate_persist_and_commit_candidate(instance, &config, &candidate, persistence).await?;

easytier-core/src/peers/attached.rs

@@ -199,7 +199,7 @@ impl AttachedPeerRuntime {
         let public_ipv6_runtime =
             CorePublicIpv6Runtime::new(runtime_config.clone(), Arc::new(()), Arc::new(()));
         let flags = peer_snapshot.flags.clone();
-        let peer_manager = Arc::new(PeerManagerCore::new(
+        let mut peer_manager = PeerManagerCore::new(
             PortablePeerManagerConfig {
                 snapshot: peer_snapshot,
                 route_algo: RouteAlgoType::Ospf,
@@ -214,7 +214,9 @@ impl AttachedPeerRuntime {
             Arc::new(()),
             None,
             Arc::new(()),
-        )?);
+        )?;
+        peer_manager.inherit_exit_nodes(&network_peer_manager);
+        let peer_manager = Arc::new(peer_manager);
         if let Err(error) = peer_manager
             .follow_network_policy(network_runtime_config, config.groups)
             .await
@@ -621,6 +623,176 @@ mod tests {
         peer_manager_with_acl(Vec::new())
     }
 
+    async fn connect_exit_peer(
+        network: &PeerManagerCore,
+        store: &CoreRuntimeConfigStore,
+        address: Ipv4Addr,
+    ) -> (Arc<PeerManagerCore>, HostPacketReceiver) {
+        let mut snapshot = store.snapshot().peer.as_ref().clone();
+        snapshot.runtime.core.routes.ipv4 = Some(IpPrefix {
+            address: address.into(),
+            prefix_len: 24,
+        });
+        if let Some(secure) = snapshot.runtime.secure_mode.as_mut() {
+            let private = StaticSecret::from([address.octets()[3]; 32]);
+            secure.local_private_key = Some(BASE64_STANDARD.encode(private.as_bytes()));
+            secure.local_public_key =
+                Some(BASE64_STANDARD.encode(PublicKey::from(&private).as_bytes()));
+        }
+        let flags = snapshot.flags.clone();
+        let store =
+            CoreRuntimeConfigStore::new(CoreRuntimeConfig::default(), Arc::new(snapshot.clone()));
+        let (sender, receiver) = host_packet_channel();
+        let public_ipv6 = CorePublicIpv6Runtime::new(store.clone(), Arc::new(()), Arc::new(()));
+        let peer = Arc::new(
+            PeerManagerCore::new(
+                PortablePeerManagerConfig {
+                    snapshot,
+                    route_algo: RouteAlgoType::Ospf,
+                    exit_nodes: Vec::new(),
+                    foreign_context_default_flags: flags,
+                },
+                Vec::new(),
+                store,
+                Arc::new(()),
+                sender,
+                public_ipv6,
+                Arc::new(()),
+                None,
+                Arc::new(()),
+            )
+            .unwrap(),
+        );
+        peer.run().await.unwrap();
+        let (server, client) = create_ring_tunnel_pair();
+        let (server_result, client_result) = tokio::join!(
+            network.add_tunnel_as_server(server),
+            peer.add_client_tunnel(client),
+        );
+        server_result.unwrap();
+        client_result.unwrap();
+        (peer, receiver)
+    }
+
+    async fn check_attached_peer_exit_nodes(secure: bool) {
+        let (network, store) = peer_manager_with_acl_and_secure(Vec::new(), secure);
+        network.run().await.unwrap();
+        let first_ip = Ipv4Addr::new(10, 82, 0, 10);
+        let second_ip = Ipv4Addr::new(10, 82, 0, 11);
+        let (first, mut first_packets) = connect_exit_peer(&network, &store, first_ip).await;
+        let (second, mut second_packets) = connect_exit_peer(&network, &store, second_ip).await;
+        network.update_exit_nodes(vec![first_ip.into()]).await;
+        let client_ip = Ipv4Addr::new(10, 82, 0, 2);
+        let attached = AttachedPeerRuntime::connect(
+            network.clone(),
+            store,
+            AttachedPeerConfig {
+                name: "exit-node-client".to_owned(),
+                virtual_ip: attached_ipv4(client_ip),
+                groups: vec!["ops".to_owned()],
+                identity_private_key: [2; 32],
+            },
+        )
+        .await
+        .unwrap();
+        for (exit, address) in [(&first, first_ip), (&second, second_ip)] {
+            wait_for_route(&attached.peer_manager, exit.my_peer_id(), address, true).await;
+            wait_for_route(exit, attached.peer_id(), client_ip, true).await;
+        }
+
+        let public_ip = Ipv4Addr::new(8, 8, 8, 8);
+        let mut request = vec![0u8; 28];
+        request[0] = 0x45;
+        request[2..4].copy_from_slice(&28u16.to_be_bytes());
+        request[8] = 64;
+        request[9] = 1;
+        request[12..16].copy_from_slice(&client_ip.octets());
+        request[16..20].copy_from_slice(&public_ip.octets());
+        request[20] = 8;
+
+        for (exit, address, packets) in [
+            (&first, first_ip, &mut first_packets),
+            (&second, second_ip, &mut second_packets),
+        ] {
+            if address == second_ip {
+                network.update_exit_nodes(vec![second_ip.into()]).await;
+            }
+            assert_eq!(
+                attached
+                    .peer_manager
+                    .get_msg_dst_peer(&public_ip.into())
+                    .await,
+                (vec![exit.my_peer_id()], true),
+            );
+            attached.send_packet(&request).await.unwrap();
+            let packet = tokio::time::timeout(Duration::from_secs(5), packets.recv())
+                .await
+                .expect("exit node did not receive the client packet")
+                .unwrap()
+                .into_core_packet();
+            assert_eq!(packet.payload(), request.as_slice());
+            let header = packet.peer_manager_header().unwrap();
+            assert!(header.is_exit_node());
+            assert_eq!(header.from_peer_id.get(), attached.peer_id());
+            assert_eq!(header.to_peer_id.get(), exit.my_peer_id());
+
+            let mut reply = request.clone();
+            reply[12..16].copy_from_slice(&public_ip.octets());
+            reply[16..20].copy_from_slice(&client_ip.octets());
+            reply[20] = 0;
+            exit.send_msg_by_ip(ZCPacket::new_with_payload(&reply), client_ip.into(), false)
+                .await
+                .unwrap();
+            let received = tokio::time::timeout(Duration::from_secs(5), attached.recv_packet())
+                .await
+                .expect("exit reply did not reach the client")
+                .unwrap();
+            assert_eq!(received.payload(), reply.as_slice());
+        }
+
+        network.update_exit_nodes(Vec::new()).await;
+        assert_eq!(
+            attached
+                .peer_manager
+                .get_msg_dst_peer(&public_ip.into())
+                .await,
+            (Vec::new(), false),
+        );
+        // Explicit mesh routes must still win, even with an exit configured.
+        network.update_exit_nodes(vec![second_ip.into()]).await;
+        assert_eq!(
+            attached
+                .peer_manager
+                .get_msg_dst_peer(&first_ip.into())
+                .await,
+            (vec![first.my_peer_id()], false),
+        );
+        request[16..20].copy_from_slice(&first_ip.octets());
+        attached.send_packet(&request).await.unwrap();
+        let packet = tokio::time::timeout(Duration::from_secs(5), first_packets.recv())
+            .await
+            .unwrap()
+            .unwrap()
+            .into_core_packet();
+        assert_eq!(packet.payload(), request.as_slice());
+        assert!(!packet.peer_manager_header().unwrap().is_exit_node());
+
+        attached.close().await;
+        first.clear_resources().await;
+        second.clear_resources().await;
+        network.clear_resources().await;
+    }
+
+    #[tokio::test]
+    async fn attached_peer_inherits_exit_nodes_and_updates() {
+        check_attached_peer_exit_nodes(false).await;
+    }
+
+    #[tokio::test]
+    async fn secure_attached_peer_inherits_exit_nodes_and_updates() {
+        check_attached_peer_exit_nodes(true).await;
+    }
+
     async fn wait_for_route(
         peer: &PeerManagerCore,
         attached_peer_id: PeerId,

easytier-core/src/peers/peer_manager.rs

@@ -1336,6 +1336,16 @@ impl PeerManagerCore {
         &self,
         config: CoreInstanceRuntimeConfig,
     ) -> anyhow::Result<Arc<CoreInstanceRuntimeConfig>> {
+        self.update_runtime_config_with(config, || ())
+            .await
+            .map(|(published, ())| published)
+    }
+
+    pub(crate) async fn update_runtime_config_with<T>(
+        &self,
+        config: CoreInstanceRuntimeConfig,
+        before_publish: impl FnOnce() -> T,
+    ) -> anyhow::Result<(Arc<CoreInstanceRuntimeConfig>, T)> {
         let _update = self.runtime_config_update.lock().await;
         let current = self.runtime_config.snapshot();
         let refresh_acl_groups = current.peer.peer_group_memberships
@@ -1345,6 +1355,9 @@ impl PeerManagerCore {
         let next_acl = reload_acl.then(|| config.services.acl.clone());
         let next_built_acl = next_acl.as_ref().map(AclRuleConfig::build).transpose()?;
 
+        // Commit prepared host resources only after all fallible validation,
+        // with no cancellation point before publishing their configuration.
+        let committed = before_publish();
         self.set_avoid_relay_data_preference(config.peer.avoid_relay_data_preference);
         let published = self
             .runtime_config
@@ -1357,7 +1370,7 @@ impl PeerManagerCore {
         if refresh_acl_groups {
             self.route.refresh_acl_groups().await;
         }
-        Ok(published)
+        Ok((published, committed))
     }
 
     /// Keeps this manager's ACL rules and group assignments aligned with a
@@ -1756,6 +1769,13 @@ impl PeerManagerCore {
         *self.exit_nodes.write().await = exit_nodes;
     }
 
+    /// Shares the network's current exit nodes, including later updates, while
+    /// keeping this peer's route lookup and packet identity independent.
+    pub(super) fn inherit_exit_nodes(&mut self, network: &Self) {
+        self.exit_nodes = network.exit_nodes.clone();
+        self.outbound_packet_router.exit_nodes = self.exit_nodes.clone();
+    }
+
     fn reload_acl(&self, acl: Option<&crate::proto::acl::Acl>) {
         self.acl_filter.reload_rules(acl);
     }

easytier-core/src/tunnel/secure_datagram.rs

@@ -1011,14 +1011,12 @@ mod tests {
         assert!(s.check_replay_for_test(1, 0, SecureDatagramDirection::AToB, now + 1));
 
         s.sync_root_key(root_key, 2, 2, true);
-        assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, now + 2));
+        // Use the actual deadline so scheduling delays before sync do not affect the test.
+        let expires_at_ms = s.sync_rx_grace_expires_at_ms.load(Ordering::Relaxed);
+        assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, expires_at_ms - 1));
+        assert!(s.check_replay_for_test(0, 1, SecureDatagramDirection::AToB, expires_at_ms - 1));
 
-        assert!(!s.check_replay_for_test(
-            0,
-            1,
-            SecureDatagramDirection::AToB,
-            now + SecureDatagramSession::SYNC_RX_GRACE_AFTER_MS + 3
-        ));
+        assert!(!s.check_replay_for_test(0, 2, SecureDatagramDirection::AToB, expires_at_ms));
     }
 
     #[test]

easytier-gui/src-tauri/tauri.windows.conf.json

@@ -7,9 +7,12 @@
             "./*.sys"
         ],
         "windows": {
+            "nsis": {
+                "installerHooks": "./windows/installer-hooks.nsh"
+            },
             "webviewInstallMode": {
                 "type": "embedBootstrapper"
             }
         }
     }
-}
+}

easytier-gui/src-tauri/windows/README.md

@@ -0,0 +1,28 @@
+# Windows installer process handling
+
+NSIS remains a current-user installer. Its helper requests elevation only
+when a running service or GUI requires it, and restores service startup and
+running state after installation. During an update it temporarily disables
+service startup because released EasyTier versions can trigger SCM recovery
+even after an explicit stop.
+
+Run `test-installer-processes.ps1` with Windows PowerShell as administrator
+on a test machine without an existing `easytier-gui` service. The test uses
+isolated executables and a service that reproduces the old exit behavior
+with failure recovery enabled. Run it with `-GuiOnly` from an unelevated
+shell to check the path that needs no elevation.
+
+## Known limitations
+
+If preparation on the welcome page has already started a helper, it is
+reused at installation time. Changing to another directory containing a
+running GUI, or relaunching the GUI between those pages, may still cause
+Tauri's default process check to fail. Rechecking those processes is a
+separate follow-up; it was recorded as a minor finding in the final review.
+
+Executable matching does not expand Windows 8.3 short-path aliases. An
+installation directory expressed using a short alias may not match the
+process image's full path. This was reproduced with LocalService's short
+temporary-directory path; the unprivileged GUI test passes when its
+temporary directory uses the full path. This is recorded as a minor
+follow-up. Use a full temporary-directory path when running the fixture.

easytier-gui/src-tauri/windows/installer-hooks.nsh

@@ -0,0 +1,95 @@
+!define EASYTIER_INSTALLER_SCRIPT "${__FILEDIR__}\installer-processes.ps1"
+
+; Tauri uninstalls the old version on its reinstall page, before PREINSTALL.
+; MUI consumes this callback on the first page (Welcome). Opening the installer
+; alone does not stop anything; the user must first leave the welcome page.
+!define MUI_PAGE_CUSTOMFUNCTION_LEAVE EasyTierPrepareInstall
+!define MUI_CUSTOMFUNCTION_GUIINIT EasyTierInstallerGuiInit
+
+Var EasyTierHelper
+Var EasyTierInstallerPid
+Var EasyTierDone
+Var EasyTierFailed
+
+!macro EasyTierPrepare RESTORE
+  ${If} $EasyTierHelper != ""
+    Return
+  ${EndIf}
+  InitPluginsDir
+  File "/oname=$PLUGINSDIR\installer-processes.ps1" "${EASYTIER_INSTALLER_SCRIPT}"
+  System::Call 'kernel32::GetCurrentProcessId() i .r0'
+  StrCpy $EasyTierInstallerPid $0
+  nsExec::ExecToStack '"$SYSDIR\WindowsPowerShell\v1.0\powershell.exe" -NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "$PLUGINSDIR\installer-processes.ps1" -InstallDir "$INSTDIR" -InstallerPid $EasyTierInstallerPid -RestoreService ${RESTORE}'
+  Pop $0
+  Pop $1
+  ${If} $0 != 0
+    MessageBox MB_OK|MB_ICONSTOP "Could not close EasyTier:$\r$\n$1" /SD IDOK
+    Abort
+  ${EndIf}
+  IntOp $1 $1 + 0
+  ${If} $1 != 0
+    System::Call 'kernel32::OpenEventW(i 0x100000, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-done") p .r0'
+    StrCpy $EasyTierDone $0
+    System::Call 'kernel32::OpenEventW(i 0x100000, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-failed") p .r0'
+    StrCpy $EasyTierFailed $0
+    ${If} $EasyTierDone == 0
+    ${OrIf} $EasyTierFailed == 0
+      MessageBox MB_OK|MB_ICONSTOP "Could not wait for EasyTier to finish updating." /SD IDOK
+      Abort
+    ${EndIf}
+    StrCpy $EasyTierHelper 1
+  ${EndIf}
+!macroend
+
+!macro EasyTierFinish
+  ${If} $EasyTierHelper != ""
+    System::Call 'kernel32::OpenEventW(i 2, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-finish") p .r0'
+    ${If} $0 != 0
+      System::Call 'kernel32::SetEvent(p r0)'
+      System::Call 'kernel32::CloseHandle(p r0)'
+    ${EndIf}
+    System::Call 'kernel32::WaitForSingleObject(p $EasyTierDone, i 65000) i .r0'
+    System::Call 'kernel32::WaitForSingleObject(p $EasyTierFailed, i 0) i .r1'
+    System::Call 'kernel32::CloseHandle(p $EasyTierDone)'
+    System::Call 'kernel32::CloseHandle(p $EasyTierFailed)'
+    StrCpy $EasyTierHelper ""
+    ${If} $0 != 0
+    ${OrIf} $1 != 258
+      MessageBox MB_OK|MB_ICONSTOP "Could not restore the EasyTier service. Please check its status in Windows Services." /SD IDOK
+      Abort
+    ${EndIf}
+  ${EndIf}
+!macroend
+
+Function EasyTierPrepareInstall
+  !insertmacro EasyTierPrepare 1
+FunctionEnd
+
+Function EasyTierInstallerGuiInit
+  ; Passive installs skip Welcome, but still run the old uninstaller page.
+  ${GetOptions} $CMDLINE "/P" $0
+  ${IfNot} ${Errors}
+    Call EasyTierPrepareInstall
+  ${EndIf}
+FunctionEnd
+
+Function un.EasyTierPrepareUninstall
+  !insertmacro EasyTierPrepare 0
+FunctionEnd
+
+!macro NSIS_HOOK_PREINSTALL
+  ; Silent installs skip all pages. Also handle a fresh/custom install path.
+  Call EasyTierPrepareInstall
+!macroend
+
+!macro NSIS_HOOK_POSTINSTALL
+  !insertmacro EasyTierFinish
+!macroend
+
+!macro NSIS_HOOK_PREUNINSTALL
+  Call un.EasyTierPrepareUninstall
+!macroend
+
+!macro NSIS_HOOK_POSTUNINSTALL
+  !insertmacro EasyTierFinish
+!macroend

easytier-gui/src-tauri/windows/installer-processes.ps1

@@ -0,0 +1,264 @@
+param(
+    [ValidateSet('Prepare', 'Run')]
+    [string]$Action = 'Prepare',
+    [Parameter(Mandatory = $true)]
+    [string]$InstallDir,
+    [Parameter(Mandatory = $true)]
+    [int]$InstallerPid,
+    [int]$RestoreService = 1
+)
+
+$ErrorActionPreference = 'Stop'
+$eventName = "Local\EasyTierInstaller-$InstallerPid"
+$exePath = [IO.Path]::GetFullPath((Join-Path $InstallDir 'easytier-gui.exe')).Replace('\\?\', '')
+Add-Type -AssemblyName System.ServiceProcess
+
+# Process.MainModule and WMI cannot reliably read an elevated GUI's path from
+# the unelevated installer. Limited query access works across that boundary.
+Add-Type @'
+using System;
+using System.ComponentModel;
+using System.Runtime.InteropServices;
+using System.Text;
+
+public static class EasyTierInstallerProcess
+{
+    [DllImport("kernel32.dll", SetLastError = true)]
+    public static extern IntPtr OpenProcess(uint access, bool inherit, int pid);
+    [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)]
+    static extern bool QueryFullProcessImageName(IntPtr process, uint flags,
+        StringBuilder path, ref int size);
+    [DllImport("kernel32.dll", SetLastError = true)]
+    static extern bool TerminateProcess(IntPtr process, uint exitCode);
+    [DllImport("kernel32.dll")]
+    static extern uint WaitForSingleObject(IntPtr handle, uint milliseconds);
+    [DllImport("kernel32.dll")]
+    public static extern bool CloseHandle(IntPtr handle);
+
+    public static string Path(IntPtr process)
+    {
+        var path = new StringBuilder(32768);
+        int size = path.Capacity;
+        if (!QueryFullProcessImageName(process, 0, path, ref size))
+        {
+            int error = Marshal.GetLastWin32Error();
+            if (WaitForSingleObject(process, 0) == 0) return null;
+            throw new Win32Exception(error);
+        }
+        return path.ToString().Replace(@"\\?\", "");
+    }
+
+    public static bool Matches(int pid, string path, bool terminate)
+    {
+        // Keep the same handle for path verification, termination and waiting,
+        // so a recycled PID cannot cause us to terminate a different process.
+        IntPtr process = OpenProcess(terminate ? 0x101001u : 0x101000u, false, pid);
+        if (process == IntPtr.Zero)
+        {
+            int error = Marshal.GetLastWin32Error();
+            if (error == 87) return false; // The process has already exited.
+            throw new Win32Exception(error);
+        }
+        try
+        {
+            if (!String.Equals(Path(process), path, StringComparison.OrdinalIgnoreCase))
+                return false;
+            if (terminate)
+            {
+                if (!TerminateProcess(process, 0) && WaitForSingleObject(process, 0) != 0)
+                    throw new Win32Exception();
+                if (WaitForSingleObject(process, 30000) != 0)
+                    throw new TimeoutException("EasyTier did not exit within 30 seconds.");
+            }
+            return true;
+        }
+        finally { CloseHandle(process); }
+    }
+}
+'@
+
+function Get-AppProcesses {
+    foreach ($process in @(Get-Process -Name easytier-gui -ErrorAction SilentlyContinue)) {
+        if ([EasyTierInstallerProcess]::Matches($process.Id, $exePath, $false)) {
+            $process
+        }
+    }
+}
+
+function Get-AppService {
+    $service = Get-WmiObject -Class Win32_Service -Filter "Name='easytier-gui'"
+    if ($service) {
+        $path = [regex]::Match($service.PathName, '^(?:"([^"]+)"|(\S+))')
+        $binary = ($path.Groups[1].Value + $path.Groups[2].Value).Replace('\\?\', '')
+        if ($binary -eq $exePath) { $service }
+    }
+}
+
+function New-InstallerEvent([string]$name) {
+    $security = New-Object Security.AccessControl.EventWaitHandleSecurity
+    # RunAs may use another administrator account. The installer keeps its
+    # original identity; only these completion signals cross the boundary.
+    foreach ($sid in @(
+        [Security.Principal.WindowsIdentity]::GetCurrent().User,
+        (New-Object Security.Principal.SecurityIdentifier 'S-1-5-32-544')
+    )) {
+        $rule = New-Object Security.AccessControl.EventWaitHandleAccessRule(
+            $sid, 'FullControl', 'Allow')
+        $security.AddAccessRule($rule)
+    }
+    $created = $false
+    New-Object Threading.EventWaitHandle(
+        $false, [Threading.EventResetMode]::ManualReset, $name, [ref]$created, $security)
+}
+
+try {
+    if ($Action -eq 'Prepare') {
+        $service = Get-AppService
+        $runningService = $service -and $service.State -ne 'Stopped'
+        $needsElevation = $runningService
+        $processes = @()
+        try {
+            $processes = @(Get-AppProcesses)
+            foreach ($process in $processes) {
+                $handle = [EasyTierInstallerProcess]::OpenProcess(0x101001, $false, $process.Id)
+                if ($handle -eq [IntPtr]::Zero) { $needsElevation = $true }
+                else { [void][EasyTierInstallerProcess]::CloseHandle($handle) }
+            }
+        }
+        catch {
+            if ($_.Exception.GetBaseException().NativeErrorCode -ne 5) { throw }
+            $needsElevation = $true
+        }
+        if (!$runningService -and !$processes.Count -and !$needsElevation) {
+            Write-Output 0
+            exit 0
+        }
+
+        $ready = New-InstallerEvent "$eventName-ready"
+        $finish = New-InstallerEvent "$eventName-finish"
+        $done = New-InstallerEvent "$eventName-done"
+        $failed = New-InstallerEvent "$eventName-failed"
+        try {
+            $arguments = '-NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass ' +
+                "-File `"$PSCommandPath`" -Action Run -InstallDir `"$InstallDir`" " +
+                "-InstallerPid $InstallerPid -RestoreService $RestoreService"
+            $start = @{
+                FilePath = "$PSHOME\powershell.exe"
+                ArgumentList = $arguments
+                WindowStyle = 'Hidden'
+                PassThru = $true
+            }
+            $identity = [Security.Principal.WindowsIdentity]::GetCurrent()
+            $principal = New-Object Security.Principal.WindowsPrincipal($identity)
+            if ($needsElevation -and !$principal.IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) {
+                $start.Verb = 'RunAs'
+            }
+            $helper = Start-Process @start
+            while (!$ready.WaitOne(100)) {
+                if ($helper.HasExited) {
+                    throw 'Could not stop EasyTier.'
+                }
+            }
+            if ($failed.WaitOne(0)) { throw 'Could not stop EasyTier.' }
+            Write-Output $helper.Id
+        }
+        finally {
+            $ready.Dispose()
+            $finish.Dispose()
+            $done.Dispose()
+            $failed.Dispose()
+        }
+    }
+    else {
+        $ready = [Threading.EventWaitHandle]::OpenExisting("$eventName-ready")
+        $finish = [Threading.EventWaitHandle]::OpenExisting("$eventName-finish")
+        $done = [Threading.EventWaitHandle]::OpenExisting("$eventName-done")
+        $failed = [Threading.EventWaitHandle]::OpenExisting("$eventName-failed")
+        $installer = [Diagnostics.Process]::GetProcessById($InstallerPid)
+        $installerExit = New-Object Threading.ManualResetEvent($false)
+        $installerExit.SafeWaitHandle = New-Object Microsoft.Win32.SafeHandles.SafeWaitHandle(
+            $installer.Handle, $false)
+        $restart = $false
+        $controller = $null
+        $serviceProcess = $null
+        $startMode = $null
+        try {
+            $service = Get-AppService
+            if ($service -and $service.State -ne 'Stopped') {
+                $controller = New-Object ServiceProcess.ServiceController('easytier-gui')
+                $restart = $RestoreService -ne 0 -and $service.State -ne 'Stop Pending' -and
+                    $service.StartMode -ne 'Disabled'
+                if ($service.StartMode -ne 'Disabled') {
+                    # Released versions can exit without reporting Stopped,
+                    # which queues SCM recovery even after an explicit stop.
+                    # Keep recovery from starting the old binary during setup.
+                    $originalStartMode = $service.StartMode
+                    $result = $service.ChangeStartMode('Disabled').ReturnValue
+                    if ($result -ne 0) { throw "Could not disable service startup: $result" }
+                    $startMode = if ($originalStartMode -eq 'Auto') { 'Automatic' } else { 'Manual' }
+                }
+                if ($service.ProcessId) {
+                    $serviceProcess = [Diagnostics.Process]::GetProcessById($service.ProcessId)
+                    # Acquire the handle before stopping, while the PID still exists.
+                    $null = $serviceProcess.Handle
+                }
+                if ($controller.Status -eq 'StartPending') {
+                    $controller.WaitForStatus('Running', [TimeSpan]::FromSeconds(30))
+                }
+                if ($controller.Status -ne 'StopPending') { $controller.Stop() }
+                $controller.WaitForStatus('Stopped', [TimeSpan]::FromSeconds(30))
+                if ($serviceProcess -and !$serviceProcess.WaitForExit(30000)) {
+                    throw 'The EasyTier service process did not exit within 30 seconds.'
+                }
+            }
+            foreach ($process in @(Get-AppProcesses)) {
+                [void][EasyTierInstallerProcess]::Matches($process.Id, $exePath, $true)
+            }
+            [void]$ready.Set()
+            [void][Threading.WaitHandle]::WaitAny(@($finish, $installerExit))
+        }
+        finally {
+            # Also restore after cancellation/failure, provided the executable
+            # still exists. A completed uninstall must not restart the service.
+            try {
+                if ($startMode) {
+                    $result = $service.ChangeStartMode($startMode).ReturnValue
+                    if ($result -ne 0) { throw "Could not restore service startup: $result" }
+                }
+                if ($restart -and (Test-Path -LiteralPath $exePath)) {
+                    $controller.Refresh()
+                    if ($controller.Status -eq 'StopPending') {
+                        $controller.WaitForStatus('Stopped', [TimeSpan]::FromSeconds(30))
+                    }
+                    if ($controller.Status -eq 'Stopped') {
+                        $controller.Start()
+                        $controller.WaitForStatus('Running', [TimeSpan]::FromSeconds(30))
+                    }
+                }
+            }
+            finally {
+                if ($controller) { $controller.Dispose() }
+                if ($serviceProcess) { $serviceProcess.Dispose() }
+                $installerExit.Dispose()
+                $installer.Dispose()
+            }
+        }
+    }
+}
+catch {
+    if ($Action -eq 'Run' -and $failed) { [void]$failed.Set() }
+    Write-Error $_ -ErrorAction Continue
+    exit 1
+}
+finally {
+    if ($Action -eq 'Run' -and $done) {
+        # Completion events remain accessible to the original user even when
+        # UAC ran the helper under a different administrator account.
+        [void]$done.Set()
+        [void]$ready.Set()
+        $ready.Dispose()
+        $finish.Dispose()
+        $done.Dispose()
+        $failed.Dispose()
+    }
+}

easytier-gui/src-tauri/windows/test-installer-processes.ps1

@@ -0,0 +1,169 @@
+# Run with Windows PowerShell as administrator on a machine without an
+# installed easytier-gui service. All processes and files are test fixtures.
+# Use -GuiOnly from an unelevated shell to test the path that needs no UAC.
+param([switch]$GuiOnly)
+
+$ErrorActionPreference = 'Stop'
+$helperScript = Join-Path $PSScriptRoot 'installer-processes.ps1'
+$testDir = Join-Path $env:TEMP ("EasyTier installer test " + [guid]::NewGuid())
+$appDir = Join-Path $testDir 'app'
+$otherDir = Join-Path $testDir 'other'
+$exe = Join-Path $appDir 'easytier-gui.exe'
+$helper = $null
+$gui = $null
+$other = $null
+$parent = $null
+$createdService = $false
+
+function Assert($condition, [string]$message) {
+    if (!$condition) { throw $message }
+}
+
+function Prepare([int]$parentPid = $PID, [int]$restore = 1) {
+    $output = & "$PSHOME\powershell.exe" -NoProfile -ExecutionPolicy Bypass `
+        -File $helperScript -InstallDir $appDir -InstallerPid $parentPid -RestoreService $restore
+    Assert ($LASTEXITCODE -eq 0) "Prepare failed: $output"
+    $helperPid = [int]($output | Select-Object -Last 1)
+    if ($helperPid) {
+        $process = [Diagnostics.Process]::GetProcessById($helperPid)
+        $null = $process.Handle
+        return $process
+    }
+}
+
+function Finish($process) {
+    if (!$process) { return }
+    $finish = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-finish")
+    $done = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-done")
+    $failed = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-failed")
+    try {
+        [void]$finish.Set()
+        Assert ($done.WaitOne(65000)) 'The helper did not signal completion.'
+        Assert (!$failed.WaitOne(0)) 'The helper reported a failure.'
+    }
+    finally {
+        $finish.Dispose()
+        $done.Dispose()
+        $failed.Dispose()
+    }
+    Assert ($process.WaitForExit(5000)) 'The helper did not finish.'
+    Assert ($process.ExitCode -eq 0) "The helper failed: $($process.ExitCode)"
+    $process.Dispose()
+}
+
+function Start-TestService {
+    Start-Service easytier-gui
+    (Get-Service easytier-gui).WaitForStatus('Running', [TimeSpan]::FromSeconds(10))
+}
+
+if (Get-Service easytier-gui -ErrorAction SilentlyContinue) {
+    throw 'An easytier-gui service already exists; refusing to touch it.'
+}
+try {
+    [void](New-Item -ItemType Directory -Path $appDir, $otherDir)
+    Add-Type -ReferencedAssemblies System.ServiceProcess -OutputAssembly $exe -OutputType ConsoleApplication @'
+using System;
+using System.ServiceProcess;
+using System.Threading;
+
+class FixtureService : ServiceBase
+{
+    static bool legacy;
+    public FixtureService() { ServiceName = "easytier-gui"; }
+    protected override void OnStop()
+    {
+        // Model released EasyTier versions that exit without reporting Stopped.
+        if (legacy) Environment.Exit(0);
+    }
+    static void Main(string[] args)
+    {
+        if (args.Length > 0)
+        {
+            legacy = args[0] == "--legacy";
+            ServiceBase.Run(new FixtureService());
+        }
+        else Thread.Sleep(Timeout.Infinite);
+    }
+}
+'@
+    Copy-Item $exe (Join-Path $otherDir 'easytier-gui.exe')
+
+    $helper = Prepare
+    Assert (!$helper) 'An idle installation must not start a helper.'
+    Write-Host 'PASS: no running process or service'
+
+    $gui = Start-Process $exe -PassThru
+    $other = Start-Process (Join-Path $otherDir 'easytier-gui.exe') -PassThru
+    $helper = Prepare
+    Assert $gui.HasExited 'The GUI was not stopped.'
+    Assert (!$other.HasExited) 'A GUI from a different installation was stopped.'
+    Finish $helper
+    $helper = $null
+    Write-Host 'PASS: close GUI, preserve a different installation'
+    if ($GuiOnly) { return }
+
+    New-Service -Name easytier-gui -BinaryPathName "`"$exe`" --legacy" -StartupType Automatic | Out-Null
+    $createdService = $true
+    & "$env:SystemRoot\System32\sc.exe" failure easytier-gui reset= 0 actions= restart/1000 | Out-Null
+    Assert ($LASTEXITCODE -eq 0) 'Could not configure service recovery.'
+    Start-TestService
+    $gui = Start-Process $exe -PassThru
+    $helper = Prepare
+    Start-Sleep -Milliseconds 1500
+    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'The service was not stopped.'
+    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Disabled') 'Service recovery was not inhibited.'
+    Assert $gui.HasExited 'The GUI alongside the service was not stopped.'
+    Move-Item $exe "$exe.old"
+    Copy-Item "$exe.old" $exe
+    Finish $helper
+    $helper = $null
+    Assert ((Get-Service easytier-gui).Status -eq 'Running') 'The running service was not restored.'
+    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Auto') 'Automatic startup was not restored.'
+    Write-Host 'PASS: stop legacy service despite failure recovery, replace executable, restore service'
+
+    Set-Service easytier-gui -StartupType Disabled
+    Stop-Service easytier-gui
+    (Get-Service easytier-gui).WaitForStatus('Stopped', [TimeSpan]::FromSeconds(10))
+    Start-Sleep -Milliseconds 1500
+    Set-Service easytier-gui -StartupType Manual
+    $helper = Prepare
+    Assert (!$helper) 'A stopped service must not start a helper.'
+    $gui = Start-Process $exe -PassThru
+    $helper = Prepare
+    Finish $helper
+    $helper = $null
+    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'A previously stopped service was started.'
+    Write-Host 'PASS: preserve stopped service state'
+
+    Start-TestService
+    $parent = Start-Process "$PSHOME\powershell.exe" -ArgumentList '-NoProfile -Command Start-Sleep 300' -WindowStyle Hidden -PassThru
+    $helper = Prepare $parent.Id
+    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'The service was not stopped before cancellation.'
+    $parent.Kill()
+    Assert ($helper.WaitForExit(35000)) 'The helper did not handle installer exit.'
+    Assert ($helper.ExitCode -eq 0) 'The helper failed while handling installer exit.'
+    $helper.Dispose()
+    $helper = $null
+    Assert ((Get-Service easytier-gui).Status -eq 'Running') 'Cancellation did not restore the service.'
+    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Manual') 'Manual startup was not restored.'
+    Write-Host 'PASS: restore service when the installer exits'
+
+    $helper = Prepare $PID 0
+    Finish $helper
+    $helper = $null
+    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'Uninstallation restarted the service.'
+    Write-Host 'PASS: do not restart the service after uninstall'
+}
+finally {
+    if ($helper -and !$helper.HasExited) { Finish $helper }
+    if ($createdService) {
+        Set-Service easytier-gui -StartupType Disabled
+        Stop-Service easytier-gui -ErrorAction SilentlyContinue
+        & "$env:SystemRoot\System32\sc.exe" delete easytier-gui | Out-Null
+    }
+    foreach ($process in @($gui, $other, $parent)) {
+        if ($process -and !$process.HasExited) { $process.Kill(); $process.WaitForExit() }
+        if ($process) { $process.Dispose() }
+    }
+    Remove-Item -LiteralPath $testDir -Recurse -Force
+}

easytier-web/src/client_manager/mod.rs

@@ -347,7 +347,11 @@ impl ClientManager {
         ) && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id)
         {
             session
-                .notify_managed_runtime_state_changed(user_id, machine_id)
+                .notify_managed_runtime_state_changed(
+                    user_id,
+                    machine_id,
+                    config_revision.as_deref(),
+                )
                 .await;
         }
         Ok(())
@@ -392,13 +396,17 @@ impl ClientManager {
                 machine_id,
                 ManagedConfigPersistedChange {
                     expected_revision: expected_config_revision,
-                    target_revision: config_revision,
+                    target_revision: config_revision.clone(),
                     dirty_instance_ids,
                 },
             );
             if changed && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id) {
                 session
-                    .notify_managed_runtime_state_changed(user_id, machine_id)
+                    .notify_managed_runtime_state_changed(
+                        user_id,
+                        machine_id,
+                        Some(&config_revision),
+                    )
                     .await;
             }
         }
@@ -416,7 +424,7 @@ impl ClientManager {
             && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id)
         {
             session
-                .notify_managed_runtime_state_changed(user_id, machine_id)
+                .notify_managed_runtime_state_changed(user_id, machine_id, None)
                 .await;
         }
     }
@@ -651,6 +659,7 @@ mod tests {
     #[derive(Debug, Clone)]
     struct TestWebhookState {
         validate_responses: Arc<tokio::sync::Mutex<VecDeque<bool>>>,
+        config_revision: Arc<tokio::sync::RwLock<String>>,
         validate_count: Arc<AtomicUsize>,
         block_second_validate: Arc<AtomicBool>,
         allow_second_validate: Arc<AtomicBool>,
@@ -665,6 +674,7 @@ mod tests {
                 validate_responses: Arc::new(tokio::sync::Mutex::new(
                     validate_responses.into_iter().collect(),
                 )),
+                config_revision: Arc::new(tokio::sync::RwLock::new("rev-initial".to_owned())),
                 validate_count: Arc::new(AtomicUsize::new(0)),
                 block_second_validate: Arc::new(AtomicBool::new(false)),
                 allow_second_validate: Arc::new(AtomicBool::new(true)),
@@ -728,8 +738,8 @@ mod tests {
 
         Json(json!({
             "valid": true,
-            "binding_version": count,
-            "config_revision": format!("validated-rev-{count}")
+            "binding_version": 1,
+            "config_revision": state.config_revision.read().await.clone()
         }))
     }
 
@@ -1270,7 +1280,8 @@ mod tests {
 
     #[tokio::test]
     async fn compact_runtime_preserves_unsupported_web_config_during_hot_patch() {
-        let (webhook_config, webhook_server, _) = test_webhook_config().await;
+        let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await;
+        *webhook_state.config_revision.write().await = "compact-initial".to_owned();
         let mut mgr = ClientManager::new(
             Db::memory_db().await,
             None,
@@ -1307,6 +1318,7 @@ mod tests {
         })
         .await;
 
+        *webhook_state.config_revision.write().await = "compact-patched".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,
@@ -1334,9 +1346,284 @@ mod tests {
         webhook_server.abort();
     }
 
+    #[tokio::test]
+    async fn receiver_takeover_preserves_running_core_until_config_arrives() {
+        for stale_config in [false, true] {
+            let (old_webhook, old_server, _) = test_webhook_config().await;
+            let mut old = ClientManager::new(
+                Db::memory_db().await,
+                None,
+                HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(),
+                Arc::new(FeatureFlags::default()),
+                old_webhook,
+            );
+            let old_addr = add_random_udp_listener(&mut old).await;
+            let machine_id = uuid::Uuid::new_v4();
+            let instance_id = uuid::Uuid::new_v4();
+            let old_user = old
+                .db()
+                .auto_create_user(MANAGED_CONFIG_TOKEN)
+                .await
+                .unwrap()
+                .id;
+            let desired = initial_managed_network_config(instance_id);
+            old.reconcile_managed_network_configs(
+                old_user,
+                machine_id,
+                vec![managed_config(instance_id, desired.clone())],
+                Some("rev-initial".to_owned()),
+                None,
+            )
+            .await
+            .unwrap();
+            let core = Arc::new(native_instance_manager());
+            let old_client = start_web_client_for_test(old_addr, machine_id, core.clone()).await;
+            wait_for_applied_revision(&old, old_user, machine_id, "rev-initial").await;
+            let original = core.instance(instance_id).unwrap();
+            drop(old_client);
+
+            let state = TestWebhookState::with_blocked_connected([true]);
+            let (webhook, server, state) = test_webhook_config_with_state(state).await;
+            let mut replacement = ClientManager::new(
+                Db::memory_db().await,
+                None,
+                HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(),
+                Arc::new(FeatureFlags::default()),
+                webhook,
+            );
+            let addr = add_random_udp_listener(&mut replacement).await;
+            let user_id = replacement
+                .db()
+                .auto_create_user(MANAGED_CONFIG_TOKEN)
+                .await
+                .unwrap()
+                .id;
+            if stale_config {
+                let obsolete_id = uuid::Uuid::new_v4();
+                replacement
+                    .reconcile_managed_network_configs(
+                        user_id,
+                        machine_id,
+                        vec![managed_config(
+                            obsolete_id,
+                            initial_managed_network_config(obsolete_id),
+                        )],
+                        Some("outdated".to_owned()),
+                        None,
+                    )
+                    .await
+                    .unwrap();
+            }
+            let _client = start_web_client_for_test(addr, machine_id, core.clone()).await;
+            wait_for_condition(
+                || async { state.connected_count() == 1 },
+                Duration::from_secs(5),
+            )
+            .await;
+            let session = replacement
+                .get_session_by_machine_id(user_id, &machine_id)
+                .unwrap();
+
+            // Authentication and real heartbeats continue while publication is
+            // withheld. Neither an empty DB nor an obsolete Exact Set owns the
+            // already running Core instance.
+            for _ in 0..3 {
+                let before = session.get_heartbeat_req().await.unwrap().report_time;
+                wait_for_condition(
+                    || async { session.get_heartbeat_req().await.unwrap().report_time != before },
+                    Duration::from_secs(3),
+                )
+                .await;
+                assert_eq!(core.instance_ids(), vec![instance_id]);
+                assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
+                assert!(session.applied_config_revision().await.is_none());
+            }
+
+            state.allow_connected();
+            // Registration must wake validation before its ten-heartbeat
+            // periodic cadence, without repeatedly registering the same binding.
+            wait_for_condition(
+                || async { state.validate_count() == 2 },
+                Duration::from_secs(2),
+            )
+            .await;
+            assert_eq!(state.connected_count(), 1);
+            assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
+
+            replacement
+                .reconcile_managed_network_configs(
+                    user_id,
+                    machine_id,
+                    vec![managed_config(instance_id, desired)],
+                    Some("rev-initial".to_owned()),
+                    None,
+                )
+                .await
+                .unwrap();
+            wait_for_applied_revision(&replacement, user_id, machine_id, "rev-initial").await;
+            assert_eq!(core.instance_ids(), vec![instance_id]);
+            assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
+
+            // A versioned authoritative empty set still removes web-owned
+            // instances and keeps the authenticated management session alive.
+            *state.config_revision.write().await = "empty".to_owned();
+            replacement
+                .reconcile_managed_network_configs(
+                    user_id,
+                    machine_id,
+                    Vec::new(),
+                    Some("empty".to_owned()),
+                    Some("rev-initial".to_owned()),
+                )
+                .await
+                .unwrap();
+            wait_for_applied_revision(&replacement, user_id, machine_id, "empty").await;
+            assert!(core.instance_ids().is_empty());
+            assert!(
+                replacement
+                    .get_session_by_machine_id(user_id, &machine_id)
+                    .is_some()
+            );
+            server.abort();
+            old_server.abort();
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+    #[ignore = "1000 real management connections; run explicitly with a raised file limit"]
+    async fn receiver_takeover_wave_allows_an_independent_mount() {
+        use futures::StreamExt;
+
+        let (old_webhook, old_server, _) = test_webhook_config().await;
+        let mut old = ClientManager::new(
+            Db::memory_db().await,
+            None,
+            HeartbeatPolicy::default(),
+            Arc::new(FeatureFlags::default()),
+            old_webhook,
+        );
+        let old_addr = add_random_udp_listener(&mut old).await;
+        let (webhook, server, state) = test_webhook_config().await;
+        let mut replacement = ClientManager::new(
+            Db::memory_db().await,
+            None,
+            HeartbeatPolicy::default(),
+            Arc::new(FeatureFlags::default()),
+            webhook,
+        );
+        let addr = add_random_udp_listener(&mut replacement).await;
+        // Provision the account once: password hashing is unrelated to a wave
+        // of existing devices reconnecting to another receiver.
+        let old_user = old
+            .db()
+            .auto_create_user(MANAGED_CONFIG_TOKEN)
+            .await
+            .unwrap()
+            .id;
+        let user_id = replacement
+            .db()
+            .auto_create_user(MANAGED_CONFIG_TOKEN)
+            .await
+            .unwrap()
+            .id;
+        let devices: Vec<_> = (0..1_000)
+            .map(|_| {
+                (
+                    uuid::Uuid::new_v4(),
+                    Arc::new(native_compact_instance_manager_with_runtime(
+                        tokio::runtime::Handle::current(),
+                    )),
+                )
+            })
+            .collect();
+        let old_clients: Vec<_> = futures::stream::iter(&devices)
+            .map(|(machine_id, core)| async {
+                old.reconcile_managed_network_configs(
+                    old_user,
+                    *machine_id,
+                    Vec::new(),
+                    Some("rev-initial".to_owned()),
+                    None,
+                )
+                .await
+                .unwrap();
+                let client = start_web_client_for_test(old_addr, *machine_id, core.clone()).await;
+                wait_for_applied_revision(&old, old_user, *machine_id, "rev-initial").await;
+                client
+            })
+            .buffer_unordered(64)
+            .collect()
+            .await;
+
+        let mount_machine = uuid::Uuid::new_v4();
+        let mount_core = Arc::new(native_compact_instance_manager_with_runtime(
+            tokio::runtime::Handle::current(),
+        ));
+        let _mount_client =
+            start_web_client_for_test(addr, mount_machine, mount_core.clone()).await;
+        wait_for_validated_user(&replacement, mount_machine).await;
+        drop(old_clients);
+        let started = tokio::time::Instant::now();
+        let takeover = futures::stream::iter(&devices)
+            .map(|(machine_id, core)| async {
+                let client = start_web_client_for_test(addr, *machine_id, core.clone()).await;
+                wait_for_validated_user(&replacement, *machine_id).await;
+                replacement
+                    .reconcile_managed_network_configs(
+                        user_id,
+                        *machine_id,
+                        Vec::new(),
+                        Some("rev-initial".to_owned()),
+                        None,
+                    )
+                    .await
+                    .unwrap();
+                wait_for_applied_revision(&replacement, user_id, *machine_id, "rev-initial").await;
+                client
+            })
+            .buffer_unordered(64)
+            .collect::<Vec<_>>();
+        let mount = async {
+            wait_for_condition(
+                || async { state.connected_count() >= 11 },
+                Duration::from_secs(12),
+            )
+            .await;
+            let started = tokio::time::Instant::now();
+            let instance_id = uuid::Uuid::new_v4();
+            replacement
+                .reconcile_managed_network_configs(
+                    user_id,
+                    mount_machine,
+                    vec![managed_config(
+                        instance_id,
+                        initial_managed_network_config(instance_id),
+                    )],
+                    Some("rev-initial".to_owned()),
+                    None,
+                )
+                .await
+                .unwrap();
+            wait_for_applied_revision(&replacement, user_id, mount_machine, "rev-initial").await;
+            assert_eq!(mount_core.instance_ids(), vec![instance_id]);
+            started.elapsed()
+        };
+        let (clients, mount_elapsed) = tokio::join!(takeover, mount);
+        assert_eq!(clients.len(), 1_000);
+        assert_eq!(replacement.list_sessions().await.len(), 1_001);
+        println!(
+            "1000 receiver takeovers: {:?}; independent mount: {:?}; validations: {}",
+            started.elapsed(),
+            mount_elapsed,
+            state.validate_count()
+        );
+        server.abort();
+        old_server.abort();
+    }
+
     #[tokio::test]
     async fn managed_web_config_revision_updates_running_core_config() {
-        let (webhook_config, webhook_server, _) = test_webhook_config().await;
+        let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await;
         let mut mgr = ClientManager::new(
             Db::memory_db().await,
             None,
@@ -1400,6 +1687,7 @@ mod tests {
 
         // Online revision update: web-owned running config is fully overwritten
         // when non-hot-patch flags such as enable_kcp_proxy change.
+        *webhook_state.config_revision.write().await = "rev-updated".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,
@@ -1454,6 +1742,7 @@ mod tests {
                 .is_none()
         );
 
+        *webhook_state.config_revision.write().await = "rev-webhook-redelivery".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,

easytier-web/src/client_manager/session.rs

@@ -133,9 +133,13 @@ pub struct SessionData {
 
     storage_token: Option<StorageToken>,
     binding_version: Option<u64>,
+    // Authentication does not prove that this receiver has the desired Exact Set.
+    // Keep that authority on the Session, not in runtime continuity shared with
+    // an earlier connection.
+    webhook_config_revision: Option<String>,
     managed_runtime: SharedManagedRuntimeState,
     direct_run_failed_instance_ids: HashSet<String>,
-    notifier: broadcast::Sender<HeartbeatRequest>,
+    notifier: broadcast::Sender<()>,
     req: Option<HeartbeatRequest>,
     location: Option<Location>,
     heartbeat_count: std::sync::atomic::AtomicU32,
@@ -165,6 +169,7 @@ impl SessionData {
             client_url,
             storage_token: None,
             binding_version: None,
+            webhook_config_revision: None,
             managed_runtime: Arc::new(Mutex::new(ManagedRuntimeState::default())),
             direct_run_failed_instance_ids: HashSet::new(),
             notifier: tx,
@@ -185,10 +190,22 @@ impl SessionData {
         self.req.clone()
     }
 
-    pub fn heartbeat_waiter(&self) -> broadcast::Receiver<HeartbeatRequest> {
+    pub fn heartbeat_waiter(&self) -> broadcast::Receiver<()> {
         self.notifier.subscribe()
     }
 
+    fn runtime_config_is_ready(&self, persisted_revision: Option<&str>) -> bool {
+        if !self.webhook_config.is_enabled() {
+            return true;
+        }
+        match self.webhook_config_revision.as_deref() {
+            // Standalone webhook integrations can omit both managed fields.
+            Some("") if self.binding_version == Some(0) => true,
+            Some(revision) if !revision.is_empty() => persisted_revision == Some(revision),
+            _ => false,
+        }
+    }
+
     pub fn location(&self) -> Option<&Location> {
         self.location.as_ref()
     }
@@ -302,6 +319,11 @@ async fn record_webhook_connected_binding_if_current(
     binding_version: u64,
 ) -> Option<ConnectedBindingRecord> {
     let session_data = session_data.upgrade()?;
+    let storage = Storage::try_from(session_data.read().await.storage.clone()).ok()?;
+    let persisted_revision = storage
+        .db()
+        .get_managed_config_revision((storage_token.user_id, storage_token.machine_id))
+        .await;
     let mut data = session_data.write().await;
     if !connection_state_matches(&data, storage_token, binding_version) {
         return Some(ConnectedBindingRecord::IdentityStale);
@@ -313,7 +335,18 @@ async fn record_webhook_connected_binding_if_current(
     {
         return Some(ConnectedBindingRecord::OwnershipLost);
     }
+    let newly_connected = data.webhook_connected_binding_version != Some(binding_version);
     data.webhook_connected_binding_version = Some(binding_version);
+    if newly_connected
+        && data.webhook_config.is_enabled()
+        && !persisted_revision
+            .as_ref()
+            .is_ok_and(|revision| data.runtime_config_is_ready(revision.as_deref()))
+    {
+        // The connected ACK makes the receiver route publishable. Retry the
+        // existing validation path now, rather than waiting ten heartbeats.
+        SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data).notify_one();
+    }
     Some(ConnectedBindingRecord::Recorded)
 }
 
@@ -748,13 +781,13 @@ impl SessionRpcService {
                     data.session_epoch,
                 );
             }
-            let runtime_notify = (authorized && data.storage_token.is_some())
-                .then(|| (data.notifier.clone(), runtime_req));
+            let runtime_notify =
+                (authorized && data.storage_token.is_some()).then(|| data.notifier.clone());
             (notify, runtime_notify)
         };
 
-        if let Some((notifier, runtime_req)) = runtime_notify {
-            let _ = notifier.send(runtime_req);
+        if let Some(notifier) = runtime_notify {
+            let _ = notifier.send(());
         }
         if let Some(notify) = notify {
             notify.notify_one();
@@ -859,7 +892,7 @@ impl SessionRpcService {
 
         let report_time = Self::heartbeat_report_timestamp(&runtime_req);
         storage.update_session_client(storage_token, report_time, true, session_epoch);
-        let _ = notifier.send(runtime_req);
+        let _ = notifier.send(());
         if let Some(notify) = validation_notify {
             notify.notify_one();
         }
@@ -1048,9 +1081,10 @@ impl Session {
         &self,
         user_id: i32,
         machine_id: uuid::Uuid,
+        published_revision: Option<&str>,
     ) {
         let notify = {
-            let data = self.data.read().await;
+            let mut data = self.data.write().await;
             if !data.auth_state.is_authorized() {
                 return;
             }
@@ -1061,10 +1095,21 @@ impl Session {
             {
                 return;
             }
-            data.req.clone().map(|req| (data.notifier.clone(), req))
+            if let Some(revision) = published_revision
+                && data.webhook_config.is_enabled()
+                && data.webhook_config_revision.as_deref() != Some(revision)
+            {
+                // Only validation establishes the desired revision. A delayed
+                // publication callback must not replace newer authority, even
+                // when its revision is still the receiver's persisted value.
+                // Fence validation already in flight and request current authority.
+                SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data)
+                    .notify_one();
+            }
+            data.req.as_ref().map(|_| data.notifier.clone())
         };
-        if let Some((notifier, req)) = notify {
-            let _ = notifier.send(req);
+        if let Some(notifier) = notify {
+            let _ = notifier.send(());
         }
     }
 
@@ -1082,10 +1127,10 @@ impl Session {
             runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1);
             runtime.runtime_config_cache_epoch = runtime.runtime_config_cache_epoch.wrapping_add(1);
             drop(runtime);
-            data.req.clone().map(|req| (data.notifier.clone(), req))
+            data.req.as_ref().map(|_| data.notifier.clone())
         };
-        if let Some((notifier, req)) = notify {
-            let _ = notifier.send(req);
+        if let Some(notifier) = notify {
+            let _ = notifier.send(());
         }
     }
 
@@ -1805,7 +1850,7 @@ mod tests {
         assert_eq!(data.binding_version, Some(1));
         assert_eq!(data.webhook_connected_binding_version, Some(1));
         drop(data);
-        assert_eq!(heartbeat_waiter.recv().await.unwrap().user_token, "token");
+        heartbeat_waiter.recv().await.unwrap();
         assert!(
             storage
                 .db()

easytier-web/src/client_manager/session/runtime_revision.rs

@@ -24,11 +24,12 @@ use crate::client_manager::{
 };
 
 async fn recv_latest_heartbeat(
-    heartbeat_waiter: &mut broadcast::Receiver<HeartbeatRequest>,
+    session_data: &std::sync::Weak<RwLock<SessionData>>,
+    heartbeat_waiter: &mut broadcast::Receiver<()>,
 ) -> Option<HeartbeatRequest> {
-    let mut req = loop {
+    loop {
         match heartbeat_waiter.recv().await {
-            Ok(req) => break req,
+            Ok(()) => break,
             Err(broadcast::error::RecvError::Lagged(skipped)) => {
                 tracing::warn!(
                     skipped,
@@ -40,21 +41,23 @@ async fn recv_latest_heartbeat(
                 return None;
             }
         }
-    };
+    }
 
     // Drop any heartbeat backlog accumulated while the previous reconcile
     // round was doing DB/RPC IO. The newest heartbeat has the freshest
     // runtime instance list, which is all this task needs.
     loop {
         match heartbeat_waiter.try_recv() {
-            Ok(next_req) => req = next_req,
+            Ok(()) => {}
             Err(broadcast::error::TryRecvError::Empty) => break,
             Err(broadcast::error::TryRecvError::Lagged(_)) => continue,
             Err(broadcast::error::TryRecvError::Closed) => return None,
         }
     }
 
-    Some(req)
+    // Notifications can be delayed by validation or publication IO. Read the
+    // current heartbeat instead of replaying the snapshot taken before that IO.
+    session_data.upgrade()?.read().await.req()
 }
 
 struct SessionClients {
@@ -65,7 +68,7 @@ struct SessionClients {
 
 pub(super) async fn reconcile_network_configs_on_heartbeat(
     session_data: std::sync::Weak<RwLock<SessionData>>,
-    mut heartbeat_waiter: broadcast::Receiver<HeartbeatRequest>,
+    mut heartbeat_waiter: broadcast::Receiver<()>,
     storage: WeakRefStorage,
     rpc_client: SessionRpcClient,
     config_client: SessionConfigClient,
@@ -78,7 +81,7 @@ pub(super) async fn reconcile_network_configs_on_heartbeat(
     };
     let mut cache = ReconcileCache::default();
     loop {
-        let Some(req) = recv_latest_heartbeat(&mut heartbeat_waiter).await else {
+        let Some(req) = recv_latest_heartbeat(&session_data, &mut heartbeat_waiter).await else {
             return;
         };
         let Some(storage) = storage.upgrade() else {
@@ -404,6 +407,15 @@ async fn prepare_reconcile_round(
             RoundStatus::Skip => return RoundStatus::Skip,
             RoundStatus::Stop => return RoundStatus::Stop,
         };
+    {
+        let Some(data) = session_data.upgrade() else {
+            return RoundStatus::Stop;
+        };
+        let data = data.read().await;
+        if !data.runtime_config_is_ready(target_config_revision.as_deref()) {
+            return RoundStatus::Skip;
+        }
+    }
     let should_apply_runtime_revision =
         target_config_revision.is_some() && target_config_revision != applied_config_revision;
     let mut scope = if should_apply_runtime_revision {
@@ -886,7 +898,9 @@ async fn begin_managed_runtime_mutation(
         return false;
     };
     let data = data.write().await;
-    if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) {
+    if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req)
+        || !data.runtime_config_is_ready(round.target_config_revision.as_deref())
+    {
         return false;
     }
     let managed_runtime = data.managed_runtime.clone();
@@ -1265,7 +1279,9 @@ async fn mark_config_revision_applied_if_current(
     };
     let notify = {
         let mut data = data.write().await;
-        if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) {
+        if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req)
+            || !data.runtime_config_is_ready(round.target_config_revision.as_deref())
+        {
             return RoundStatus::Ready(());
         }
         record_applied_config_revision(
@@ -1319,6 +1335,286 @@ mod tests {
 
     use super::*;
 
+    async fn takeover_session(
+        persisted_revision: Option<&str>,
+    ) -> (
+        crate::client_manager::storage::Storage,
+        std::sync::Arc<RwLock<SessionData>>,
+        ReconcileRound,
+    ) {
+        use super::super::{SessionAuthState, webhook_validation};
+        use crate::client_manager::storage::{Storage, StorageToken};
+
+        let storage = Storage::new(crate::db::Db::memory_db().await);
+        let user_id = storage.db().auto_create_user("token").await.unwrap().id;
+        let machine_id = uuid::Uuid::new_v4();
+        if let Some(revision) = persisted_revision {
+            storage
+                .db()
+                .set_managed_config_revision((user_id, machine_id), revision)
+                .await
+                .unwrap();
+        }
+        let req = HeartbeatRequest {
+            user_token: "token".to_owned(),
+            machine_id: Some(machine_id.into()),
+            running_network_instances: vec![uuid::Uuid::new_v4().into()],
+            ..Default::default()
+        };
+        let client_url = "tcp://127.0.0.1:1001".parse().unwrap();
+        let webhook = std::sync::Arc::new(crate::webhook::WebhookConfig::new(
+            Some("http://127.0.0.1:1".to_owned()),
+            None,
+            None,
+            None,
+            None,
+        ));
+        let mut data = SessionData::new(
+            storage.weak_ref(),
+            client_url,
+            None,
+            std::sync::Arc::new(crate::FeatureFlags::default()),
+            webhook.clone(),
+        );
+        data.req = Some(req.clone());
+        data.auth_state = SessionAuthState::Authorized;
+        data.binding_version = Some(1);
+        data.webhook_connected_binding_version = Some(1);
+        data.storage_token = Some(StorageToken {
+            token: req.user_token.clone(),
+            client_url: data.client_url.clone(),
+            machine_id,
+            user_id,
+        });
+        data.managed_runtime = storage.bind_managed_runtime_state(user_id, machine_id, None, 0);
+        data.managed_runtime().applied_config_revision = persisted_revision.map(str::to_owned);
+        let session_data = std::sync::Arc::new(RwLock::new(data));
+        let client_url = session_data.read().await.client_url.clone();
+        webhook_validation::apply_success(
+            &std::sync::Arc::downgrade(&session_data),
+            webhook_validation::WebhookValidationInput {
+                storage: storage.clone(),
+                webhook_config: webhook,
+                client_url,
+                applied_config_revision: persisted_revision.map(str::to_owned),
+                applied_config_revision_known: false,
+                failed_instance_ids: Vec::new(),
+                req: req.clone(),
+                machine_id,
+            },
+            webhook_validation::WebhookHeartbeatValidation {
+                config_revision: "current".to_owned(),
+                binding_version: 1,
+            },
+            user_id,
+            0,
+        )
+        .await;
+        let mut round = round_with_revision_state(persisted_revision, false);
+        round.req = req;
+        round.machine_id = machine_id;
+        round.user_id = user_id;
+        round.runtime_config_epoch = session_data
+            .read()
+            .await
+            .managed_runtime()
+            .runtime_config_epoch;
+        (storage, session_data, round)
+    }
+
+    #[tokio::test]
+    async fn takeover_skips_missing_and_outdated_receiver_config() {
+        for revision in [None, Some("outdated")] {
+            let (storage, data, round) = takeover_session(revision).await;
+            let rpc = easytier::proto::rpc::bidirect::BidirectRpcManager::new();
+            let mut client: SessionRpcClient = rpc.rpc_client().scoped_client::<
+                easytier::proto::api::manage::WebClientServiceClientFactory<BaseController>,
+            >(1, 1, String::new());
+
+            assert!(matches!(
+                prepare_reconcile_round(
+                    &std::sync::Arc::downgrade(&data),
+                    &storage.weak_ref().upgrade().unwrap(),
+                    &mut client,
+                    round.req,
+                )
+                .await,
+                RoundStatus::Skip
+            ));
+        }
+    }
+
+    #[tokio::test]
+    async fn takeover_fences_mutations_and_applied_reports_from_unready_rounds() {
+        for revision in [None, Some("outdated")] {
+            let (storage, data, mut round) = takeover_session(revision).await;
+            let weak = std::sync::Arc::downgrade(&data);
+            let mut mutation = RuntimeMutationFence::default();
+            assert!(!begin_managed_runtime_mutation(&weak, &round, &mut mutation).await);
+            assert!(!mutation.started);
+
+            round.should_apply_runtime_revision = true;
+            mark_config_revision_applied_if_current(
+                &weak,
+                &storage.weak_ref().upgrade().unwrap(),
+                &round,
+                &ReconcileOutcome::default(),
+                &mutation,
+            )
+            .await;
+            let data = data.read().await;
+            let runtime = data.managed_runtime();
+            assert!(!runtime.applied_config_revision_known);
+            assert_eq!(runtime.applied_config_revision.as_deref(), revision);
+        }
+    }
+
+    async fn validate_takeover_session(
+        storage: &crate::client_manager::storage::Storage,
+        data: &std::sync::Arc<RwLock<SessionData>>,
+        round: &ReconcileRound,
+        revision: &str,
+        epoch: u64,
+    ) {
+        use super::super::webhook_validation;
+
+        let (webhook_config, client_url) = {
+            let data = data.read().await;
+            (data.webhook_config.clone(), data.client_url.clone())
+        };
+        webhook_validation::apply_success(
+            &std::sync::Arc::downgrade(data),
+            webhook_validation::WebhookValidationInput {
+                storage: storage.clone(),
+                webhook_config,
+                client_url,
+                applied_config_revision: None,
+                applied_config_revision_known: false,
+                failed_instance_ids: Vec::new(),
+                req: round.req.clone(),
+                machine_id: round.machine_id,
+            },
+            webhook_validation::WebhookHeartbeatValidation {
+                config_revision: revision.to_owned(),
+                binding_version: 1,
+            },
+            round.user_id,
+            epoch,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn takeover_new_publication_fences_late_validation_and_publication_callbacks() {
+        use super::super::{HeartbeatPolicy, Session};
+        use std::sync::Arc;
+
+        let (storage, data, round) = takeover_session(Some("current")).await;
+        let mut session = Session::new(
+            storage.weak_ref(),
+            "tcp://127.0.0.1:1001".parse().unwrap(),
+            None,
+            HeartbeatPolicy::default(),
+            Arc::new(crate::FeatureFlags::default()),
+            data.read().await.webhook_config.clone(),
+            0,
+        );
+        session.data = data.clone();
+        let old_epoch = data.read().await.webhook_validation_change_epoch;
+        storage
+            .db()
+            .set_managed_config_revision((round.user_id, round.machine_id), "newer")
+            .await
+            .unwrap();
+        storage.record_full_managed_config_change(round.user_id, round.machine_id, Some("newer"));
+        session
+            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer"))
+            .await;
+
+        validate_takeover_session(&storage, &data, &round, "current", old_epoch).await;
+        session
+            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("current"))
+            .await;
+        let current_epoch = {
+            let data = data.read().await;
+            assert_eq!(data.webhook_config_revision.as_deref(), Some("current"));
+            assert!(!data.runtime_config_is_ready(Some("newer")));
+            assert!(data.webhook_validation_dirty);
+            data.webhook_validation_change_epoch
+        };
+        assert!(
+            !begin_managed_runtime_mutation(
+                &Arc::downgrade(&data),
+                &round,
+                &mut RuntimeMutationFence::default()
+            )
+            .await
+        );
+
+        validate_takeover_session(&storage, &data, &round, "newer", current_epoch).await;
+        assert!(data.read().await.runtime_config_is_ready(Some("newer")));
+
+        // A newer Console revision may be observed after a publication commits
+        // but before its callback runs. SQLite still containing that publication
+        // does not allow the late callback to replace the newer authority.
+        validate_takeover_session(&storage, &data, &round, "newest", current_epoch).await;
+        session
+            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer"))
+            .await;
+        let data = data.read().await;
+        assert_eq!(data.webhook_config_revision.as_deref(), Some("newest"));
+        assert!(!data.runtime_config_is_ready(Some("newer")));
+        assert!(data.webhook_validation_dirty);
+    }
+
+    #[tokio::test]
+    async fn unchanged_validation_does_not_replay_an_old_heartbeat() {
+        let (storage, data, round) = takeover_session(Some("current")).await;
+        let mut notifications = data.read().await.heartbeat_waiter();
+        let epoch = data.read().await.webhook_validation_change_epoch;
+        validate_takeover_session(&storage, &data, &round, "current", epoch).await;
+        assert!(matches!(
+            notifications.try_recv(),
+            Err(broadcast::error::TryRecvError::Empty)
+        ));
+
+        let newest_instance = uuid::Uuid::new_v4();
+        {
+            let mut data = data.write().await;
+            data.notifier.send(()).unwrap();
+            data.req.as_mut().unwrap().running_network_instances = vec![newest_instance.into()];
+        }
+        let req = recv_latest_heartbeat(&std::sync::Arc::downgrade(&data), &mut notifications)
+            .await
+            .unwrap();
+        assert_eq!(req.running_network_instances, vec![newest_instance.into()]);
+    }
+
+    #[tokio::test]
+    async fn authoritative_empty_config_remains_ready_but_missing_revision_does_not() {
+        let (storage, data, round) = takeover_session(Some("current")).await;
+        let mut mutation = RuntimeMutationFence::default();
+        assert!(round.local_configs.is_empty());
+        assert!(
+            begin_managed_runtime_mutation(
+                &std::sync::Arc::downgrade(&data),
+                &round,
+                &mut mutation
+            )
+            .await
+        );
+        let epoch = data.read().await.webhook_validation_change_epoch;
+        validate_takeover_session(&storage, &data, &round, "", epoch).await;
+        let mut data = data.write().await;
+        assert!(!data.runtime_config_is_ready(None));
+        assert!(!data.runtime_config_is_ready(Some("current")));
+        data.binding_version = Some(0);
+        assert!(
+            data.runtime_config_is_ready(None),
+            "legacy webhook without managed fields"
+        );
+    }
+
     fn config_with_port_forwards(port_forwards: Vec<PortForwardConfig>) -> NetworkConfig {
         NetworkConfig {
             instance_id: Some("11111111-1111-1111-1111-111111111111".to_string()),

easytier-web/src/client_manager/session/webhook_validation.rs

@@ -393,7 +393,7 @@ pub(super) async fn apply_success(
     validation_change_epoch: u64,
 ) {
     let WebhookHeartbeatValidation {
-        config_revision: _,
+        config_revision,
         binding_version,
     } = validation;
 
@@ -408,6 +408,7 @@ pub(super) async fn apply_success(
         validation_notify,
         runtime_req,
         session_epoch,
+        notify_runtime,
     ) = {
         let mut data = session_data.write().await;
         let Some(runtime_req) = data.req.clone() else {
@@ -436,6 +437,9 @@ pub(super) async fn apply_success(
             return;
         }
 
+        let notify_runtime = !data.auth_state.is_authorized()
+            || data.webhook_config_revision.as_deref() != Some(config_revision.as_str());
+        let runtime_changed = notify_runtime || data.binding_version != Some(binding_version);
         let previous_connected_binding_version = data.webhook_connected_binding_version;
         let client_url = data.client_url.clone();
         let is_new_storage_token = data.storage_token.is_none();
@@ -461,6 +465,11 @@ pub(super) async fn apply_success(
         let storage_token = storage_token.clone();
         data.auth_state = SessionAuthState::Authorized;
         data.binding_version = Some(binding_version);
+        data.webhook_config_revision = Some(config_revision);
+        if runtime_changed {
+            let mut runtime = data.managed_runtime();
+            runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1);
+        }
         if is_new_storage_token {
             tracing::info!(
                 machine_id = %input.machine_id,
@@ -520,6 +529,7 @@ pub(super) async fn apply_success(
             validation_notify,
             runtime_req,
             data.session_epoch,
+            notify_runtime,
         )
     };
 
@@ -541,7 +551,9 @@ pub(super) async fn apply_success(
         .await;
     }
 
-    let _ = notifier.send(runtime_req);
+    if notify_runtime {
+        let _ = notifier.send(());
+    }
 }
 
 async fn wait_webhook_connection_transition(
@@ -563,6 +575,87 @@ async fn wait_webhook_connection_transition(
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn unchanged_validation_does_not_repeat_runtime_notifications() {
+        let storage = Storage::new(crate::db::Db::memory_db().await);
+        let user_id = storage.db().auto_create_user("token").await.unwrap().id;
+        let machine_id = uuid::Uuid::new_v4();
+        let session_data = validation_session(machine_id).await;
+        {
+            let mut data = session_data.write().await;
+            data.storage = storage.weak_ref();
+            data.auth_state = SessionAuthState::Init;
+        }
+        let mut notifications = session_data.read().await.heartbeat_waiter();
+        for attempt in 0..100 {
+            let (input, epoch) = {
+                let data = session_data.read().await;
+                (
+                    WebhookValidationInput {
+                        storage: storage.clone(),
+                        webhook_config: data.webhook_config.clone(),
+                        client_url: data.client_url.clone(),
+                        applied_config_revision: (attempt > 0).then(|| "current".to_owned()),
+                        applied_config_revision_known: attempt > 0,
+                        failed_instance_ids: Vec::new(),
+                        req: data.req.clone().unwrap(),
+                        machine_id,
+                    },
+                    data.webhook_validation_change_epoch,
+                )
+            };
+            apply_success(
+                &Arc::downgrade(&session_data),
+                input,
+                WebhookHeartbeatValidation {
+                    config_revision: "current".to_owned(),
+                    binding_version: if attempt == 99 { 2 } else { 1 },
+                },
+                user_id,
+                epoch,
+            )
+            .await;
+            if attempt == 0 {
+                notifications
+                    .try_recv()
+                    .expect("initial authentication wakes runtime");
+                let data = session_data.read().await;
+                let mut runtime = data.managed_runtime();
+                runtime.applied_config_revision = Some("current".to_owned());
+                runtime.applied_config_revision_known = true;
+            } else {
+                assert!(matches!(
+                    notifications.try_recv(),
+                    Err(tokio::sync::broadcast::error::TryRecvError::Empty)
+                ));
+            }
+        }
+        // A fresh heartbeat still wakes the reconciler after repeated success.
+        let instance_id = uuid::Uuid::new_v4();
+        let mut fresh = session_data.read().await.req.clone().unwrap();
+        fresh.running_network_instances = vec![instance_id.into()];
+        SessionRpcService {
+            data: session_data.clone(),
+            heartbeat_policy: super::super::HeartbeatPolicy::default(),
+        }
+        .handle_heartbeat(fresh)
+        .await
+        .unwrap();
+        notifications
+            .try_recv()
+            .expect("fresh heartbeat wakes runtime");
+        assert_eq!(
+            session_data
+                .read()
+                .await
+                .req
+                .as_ref()
+                .unwrap()
+                .running_network_instances,
+            vec![instance_id.into()]
+        );
+    }
+
     async fn validation_session(machine_id: uuid::Uuid) -> Arc<RwLock<SessionData>> {
         let storage = Storage::new(crate::db::Db::memory_db().await);
         let mut data = SessionData::new(

easytier/src/core.rs

@@ -19,14 +19,17 @@ use crate::{
 };
 use anyhow::Context;
 use cidr::IpCidr;
-use clap::{Arg, ArgMatches, CommandFactory, FromArgMatches, Parser};
+use clap::{
+    Arg, ArgMatches, CommandFactory, FromArgMatches, Parser, builder::TypedValueParser as _,
+};
 use easytier_core::config::{DhcpConfig, EncryptionAlgorithm, normalize_secure_mode_config};
 use guarden::defer;
 use prost_types::field_descriptor_proto::Type;
 use rust_i18n::t;
 use serde_json::{Value, json};
-use std::str::{FromStr, ParseBoolError};
+use std::str::FromStr;
 use std::{
+    fmt::Write as _,
     net::{IpAddr, SocketAddr},
     path::PathBuf,
     process::ExitCode,
@@ -1018,8 +1021,8 @@ fn win_service_event_loop(
 
     let normal_status = ServiceStatus {
         service_type: ServiceType::OWN_PROCESS,
-        current_state: ServiceState::Running,
-        controls_accepted: ServiceControlAccept::STOP,
+        current_state: ServiceState::Stopped,
+        controls_accepted: ServiceControlAccept::empty(),
         exit_code: ServiceExitCode::Win32(0),
         checkpoint: 0,
         wait_hint: Duration::default(),
@@ -1102,9 +1105,9 @@ fn cli_command() -> clap::Command {
         value.parse::<T>().map(|value| json!(value))
     }
 
-    fn negated(value: &str) -> Result<Value, ParseBoolError> {
-        value.parse::<bool>().map(|value| json!(!value))
-    }
+    let parser = clap::builder::BoolishValueParser::new();
+    let boolish = parser.map(|v| json!(v));
+    let negated = parser.map(|v| json!(!v));
 
     Cli::command().args(Flags::flags().iter().map(|field| {
         let name = field.name();
@@ -1115,11 +1118,11 @@ fn cli_command() -> clap::Command {
                 .short('u')
                 .num_args(0..=1)
                 .default_missing_value("true")
-                .value_parser(negated),
+                .value_parser(negated.clone()),
             "enable_ipv6" => arg(name, "disable_ipv6")
                 .num_args(0..=1)
                 .default_missing_value("true")
-                .value_parser(negated),
+                .value_parser(negated.clone()),
             // The command line names the algorithm instead of spelling the field.
             // The values are the schema's own, which the patch reads.
             "data_compress_algo" => {
@@ -1138,7 +1141,7 @@ fn cli_command() -> clap::Command {
                     Type::Bool => arg
                         .num_args(0..=1)
                         .default_missing_value("true")
-                        .value_parser(parsed::<bool>),
+                        .value_parser(boolish.clone()),
                     Type::Uint32 => arg.value_parser(parsed::<u32>),
                     Type::Uint64 => arg.value_parser(parsed::<u64>),
                     // An enum flag takes the name the schema spells, and the patch's
@@ -1209,7 +1212,11 @@ fn win_service_main(arg: Vec<std::ffi::OsString>) {
 
 async fn run_main(cli: Cli) -> anyhow::Result<()> {
     defer!(dump_profile(0););
-    log::init(&cli.logging_options, true)?;
+    log::init_with_default_console_targets(
+        &cli.logging_options,
+        true,
+        &["CORE", "easytier_core::instance::manager"],
+    )?;
 
     let manager = Arc::new(native_cli_instance_manager().with_config_path(cli.config_dir.clone()));
 
@@ -1358,10 +1365,17 @@ async fn run_main(cli: Cli) -> anyhow::Result<()> {
     tokio::select! {
         _ = manager.wait() => {
             let infos = manager.collect_network_infos().await?;
-            if infos
-                .into_values()
-                .filter_map(|info| info.error_msg).next().is_some() {
-                return Err(anyhow::anyhow!("some instances stopped with errors"));
+            let mut errors = String::new();
+            for (instance_id, info) in infos {
+                if let Some(error) = info.error_msg {
+                    if errors.is_empty() {
+                        errors.push_str("some instances stopped with errors:");
+                    }
+                    write!(errors, "\n{instance_id}: {error}").unwrap();
+                }
+            }
+            if !errors.is_empty() {
+                return Err(anyhow::anyhow!(errors));
             }
         }
         _ = tokio::signal::ctrl_c() => {
@@ -1487,7 +1501,7 @@ pub async fn main() -> ExitCode {
     let mut ret_code = 0;
 
     if let Err(error) = run_main(cli).await {
-        log::error!(%error);
+        log::error!(error = %format_args!("{error:#}"));
         ret_code = 1;
     }
 
@@ -1772,6 +1786,44 @@ enabled = true
         let patch = parse_flags(&["easytier", "--disable-ipv6"]).unwrap();
         assert_eq!(patch.enable_ipv6, Some(false));
 
+        // Boolish tests: accepts 1/0, yes/no, on/off, true/false
+        assert_eq!(
+            parse_flags(&["easytier", "--latency-first", "1"])
+                .unwrap()
+                .latency_first,
+            Some(true)
+        );
+        assert_eq!(
+            parse_flags(&["easytier", "--latency-first", "0"])
+                .unwrap()
+                .latency_first,
+            Some(false)
+        );
+        assert_eq!(
+            parse_flags(&["easytier", "--latency-first", "yes"])
+                .unwrap()
+                .latency_first,
+            Some(true)
+        );
+        assert_eq!(
+            parse_flags(&["easytier", "--latency-first", "no"])
+                .unwrap()
+                .latency_first,
+            Some(false)
+        );
+        assert_eq!(
+            parse_flags(&["easytier", "--disable-encryption", "1"])
+                .unwrap()
+                .enable_encryption,
+            Some(false)
+        );
+        assert_eq!(
+            parse_flags(&["easytier", "--disable-encryption", "0"])
+                .unwrap()
+                .enable_encryption,
+            Some(true)
+        );
+
         // Compression tests: case-insensitive, accepts None/none/zstd/Zstd/ZSTD, rejects invalid
         let patch = parse_flags(&["easytier", "--compression", "none"]).unwrap();
         assert_eq!(

easytier/src/socket/fake_tcp/netfilter/linux_bpf.rs

@@ -870,20 +870,28 @@ mod tests {
             let suffix = rand::thread_rng().gen_range(10000..99999);
             let capture_name = format!("elbc{suffix}");
             let sender_name = format!("elbs{suffix}");
+            // Assign MACs at creation so udev does not replace them after readback.
             run_ip(&[
                 "link",
                 "add",
                 &capture_name,
+                "address",
+                "02:00:00:00:00:01",
                 "type",
                 "veth",
                 "peer",
                 "name",
                 &sender_name,
+                "address",
+                "02:00:00:00:00:02",
             ])?;
 
             let setup_result = (|| {
                 run_ip(&["link", "set", "dev", &capture_name, "up"])?;
                 run_ip(&["link", "set", "dev", &sender_name, "up"])?;
+                // A single-device GETLINK synchronizes pending linkwatch work,
+                // activating the first endpoint's TX queue before any test sends.
+                run_ip(&["link", "show", "dev", &capture_name])?;
                 let interfaces = datalink::interfaces();
                 let find_mac = |name: &str| {
                     interfaces

easytier/src/socket/tcp.rs

@@ -303,7 +303,14 @@ pub(crate) async fn bind_tcp_listener(
     let addr = bind_options.local_addr.ok_or_else(|| {
         TunnelError::InvalidAddr("tcp listener requires a local bind address".to_owned())
     })?;
-    if bind_options.bind_device.is_none() && purpose == TcpListenPurpose::PortLease {
+    if bind_options.bind_device.is_none()
+        && matches!(
+            purpose,
+            TcpListenPurpose::PortLease | TcpListenPurpose::PortForward
+        )
+    {
+        // Port forwards follow the bind IP across interface recreation. An
+        // implicit device binding would retain the deleted interface's index.
         bind_options.bind_device = Some(String::new());
     }
     let listener = create_tcp_socket(addr, &bind_options).await?.listen(1024)?;
@@ -363,6 +370,22 @@ mod tests {
 
     use super::*;
 
+    #[cfg(target_os = "linux")]
+    #[rstest::rstest]
+    #[tokio::test]
+    async fn port_forward_listener_honors_device_binding(
+        #[values(None, Some("lo"), Some(""))] device: Option<&str>,
+    ) {
+        let mut options = TcpListenOptions::port_forward("127.0.0.1:0".parse().unwrap());
+        options.bind.bind_device = device.map(str::to_owned);
+        let listener = bind_tcp_listener(options).await.unwrap();
+        let bound = SockRef::from(&listener.listener).device().unwrap();
+        assert_eq!(
+            bound.as_deref(),
+            device.filter(|name| !name.is_empty()).map(str::as_bytes)
+        );
+    }
+
     #[tokio::test]
     async fn runtime_tcp_owned_split_preserves_full_duplex_and_shutdown() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

easytier/src/socket/udp.rs

@@ -146,27 +146,17 @@ impl RuntimeUdpSocketFactory {
             return BindDev::from(bind_device.as_str());
         }
 
+        // Port forwards bind only their local address by default, so replacing
+        // its interface does not strand the socket on the old interface index.
         if matches!(
             options.purpose,
-            UdpSocketPurpose::DirectConnect
-                | UdpSocketPurpose::PortBoundListener
-                | UdpSocketPurpose::PortForward
+            UdpSocketPurpose::DirectConnect | UdpSocketPurpose::PortBoundListener
         ) {
             return BindDev::Auto;
         }
 
         BindDev::Disabled
     }
-
-    fn reuse_addr_for(&self, options: &UdpBindOptions) -> bool {
-        options.reuse_addr
-            || (matches!(
-                options.purpose,
-                UdpSocketPurpose::PortBoundListener
-                    | UdpSocketPurpose::ProxyNat
-                    | UdpSocketPurpose::PortForward
-            ) && !cfg!(target_os = "windows"))
-    }
 }
 
 /// Creates a host-owned UDP socket, including any required protection, before
@@ -181,7 +171,9 @@ pub(crate) async fn create_udp_socket(options: &UdpBindOptions) -> anyhow::Resul
         .dev(factory.bind_device_for(options))
         .net_ns(NetNS::from_socket_context(&options.context))
         .only_v6(options.only_v6)
-        .reuse_addr(factory.reuse_addr_for(options))
+        // Independent UDP receivers must not share a bound endpoint, including
+        // kernel-assigned ports. Only an explicit request may enable reuse.
+        .reuse_addr(options.reuse_addr)
         .reuse_port(options.reuse_port)
         .maybe_socket_mark(options.context.socket_mark)
         .need_protect(options.need_protect)
@@ -216,6 +208,78 @@ mod tests {
 
     use super::*;
 
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn runtime_udp_bind_rejects_occupied_port() {
+        for local_addr in ["0.0.0.0:0", "[::]:0"] {
+            let local_addr = local_addr.parse().unwrap();
+            for bind in [
+                UdpBindOptions::port_bound_listener(local_addr),
+                UdpBindOptions::proxy_nat().with_local_addr(Some(local_addr)),
+                UdpBindOptions::port_forward(local_addr),
+            ] {
+                let bind = bind.with_only_v6(true);
+                let listener = create_udp_socket(&bind).await.unwrap();
+                let occupied = bind.with_local_addr(Some(listener.local_addr().unwrap()));
+
+                for reuse_addr in [false, true] {
+                    let error = create_udp_socket(&occupied.clone().with_reuse_addr(reuse_addr))
+                        .await
+                        .expect_err("another socket must not share a bound receive endpoint");
+                    assert!(matches!(
+                        error.downcast_ref::<easytier_core::tunnel::TunnelError>(),
+                        Some(easytier_core::tunnel::TunnelError::IOError(error))
+                            if error.kind() == std::io::ErrorKind::AddrInUse
+                    ));
+                }
+
+                drop(listener);
+                create_udp_socket(&occupied).await.unwrap();
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn runtime_udp_listeners_keep_ephemeral_ports_and_packets_separate() {
+        for (bind_addr, loopback) in [
+            ("0.0.0.0:0", "127.0.0.1".parse::<IpAddr>().unwrap()),
+            ("[::]:0", "::1".parse::<IpAddr>().unwrap()),
+        ] {
+            let bind_addr = bind_addr.parse().unwrap();
+            let binds = [
+                UdpBindOptions::port_bound_listener(bind_addr),
+                UdpBindOptions::proxy_nat().with_local_addr(Some(bind_addr)),
+                UdpBindOptions::port_forward(bind_addr),
+            ];
+            let mut listeners = Vec::new();
+            let mut ports = std::collections::HashSet::new();
+            for index in 0..128 {
+                let bind = binds[index % binds.len()].clone().with_only_v6(true);
+                let listener = create_udp_socket(&bind).await.unwrap();
+                assert!(ports.insert(listener.local_addr().unwrap().port()));
+                listeners.push(listener);
+            }
+
+            let sender = UdpSocket::bind(SocketAddr::new(loopback, 0)).await.unwrap();
+            for (index, listener) in listeners.iter().enumerate() {
+                let destination = SocketAddr::new(loopback, listener.local_addr().unwrap().port());
+                sender.send_to(&[index as u8], destination).await.unwrap();
+            }
+            for (index, listener) in listeners.iter().enumerate() {
+                let mut packet = [0; 16];
+                let (len, source) = tokio::time::timeout(
+                    std::time::Duration::from_secs(2),
+                    listener.recv_from(&mut packet),
+                )
+                .await
+                .expect("listener did not receive its packet")
+                .unwrap();
+                assert_eq!(&packet[..len], &[index as u8]);
+                assert_eq!(source, sender.local_addr().unwrap());
+            }
+        }
+    }
+
     #[cfg(any(target_os = "linux", target_os = "android"))]
     #[tokio::test]
     async fn runtime_udp_socket_reports_ipv4_destination_ip() {
@@ -360,7 +424,7 @@ mod tests {
     }
 
     #[test]
-    fn factory_interprets_bind_defaults_by_purpose() {
+    fn factory_interprets_bind_device_by_purpose() {
         let listener_addr = SocketAddr::from(([0, 0, 0, 0], 11010));
         let factory = RuntimeUdpSocketFactory::new();
 
@@ -374,7 +438,7 @@ mod tests {
         ));
         assert!(matches!(
             factory.bind_device_for(&UdpBindOptions::port_forward(listener_addr)),
-            BindDev::Auto
+            BindDev::Disabled
         ));
         assert!(matches!(
             factory.bind_device_for(&UdpBindOptions::port_lease(listener_addr)),
@@ -384,28 +448,18 @@ mod tests {
             factory.bind_device_for(&UdpBindOptions::hole_punch_control()),
             BindDev::Disabled
         ));
-        assert_eq!(
-            factory.reuse_addr_for(&UdpBindOptions::port_bound_listener(listener_addr)),
-            !cfg!(target_os = "windows")
-        );
-        assert!(!factory.reuse_addr_for(&UdpBindOptions::hole_punch_control()));
-        assert_eq!(
-            factory.reuse_addr_for(&UdpBindOptions::proxy_nat()),
-            !cfg!(target_os = "windows")
-        );
-        assert_eq!(
-            factory.reuse_addr_for(&UdpBindOptions::port_forward(listener_addr)),
-            !cfg!(target_os = "windows")
-        );
-        assert!(!factory.reuse_addr_for(&UdpBindOptions::port_lease(listener_addr)));
     }
 
-    #[test]
-    fn factory_applies_listener_bind_device_option() {
+    #[rstest::rstest]
+    fn factory_applies_listener_bind_device_option(#[values(false, true)] port_forward: bool) {
         let listener_addr = SocketAddr::from(([0, 0, 0, 0], 11010));
         let factory = RuntimeUdpSocketFactory::new();
-        let options = UdpBindOptions::port_bound_listener(listener_addr)
-            .with_bind_device(Some("eth0".to_owned()));
+        let options = if port_forward {
+            UdpBindOptions::port_forward(listener_addr)
+        } else {
+            UdpBindOptions::port_bound_listener(listener_addr)
+        }
+        .with_bind_device(Some("eth0".to_owned()));
 
         match factory.bind_device_for(&options) {
             BindDev::Custom(dev) => assert_eq!(dev, "eth0"),

easytier/src/tests/shared_virtual_nic.rs

@@ -554,18 +554,24 @@ fn shared_tun_claim_cleanup_outlives_member_runtime() {
 #[case::static_ipv4(false)]
 #[case::with_dhcpv6(true)]
 #[tokio::test]
-#[serial_test::serial]
 async fn same_namespace_members_share_tun_across_independent_networks(#[case] dhcpv6: bool) {
     let dev_name = test_dev_name();
-    let first_peer_ns = TestNetnsGuard::new(short_name("sva"), "10.231.1.2/24", "fd31::2/64");
-    let second_peer_ns = TestNetnsGuard::new(short_name("svb"), "10.231.2.2/24", "fd32::2/64");
+    let mut network = TestNetwork::new();
+    // Nextest runs each case in its own process, so their shared TUNs must
+    // also have separate namespaces to avoid competing routes for the same IPs.
+    network.create_empty_namespace("shared");
+    network.create_namespace("first_peer", "10.231.1.2/24", "fd31::2/64");
+    network.create_namespace("second_peer", "10.231.2.2/24", "fd32::2/64");
+    let shared_ns = network.namespace("shared");
+    let first_peer_ns = network.namespace("first_peer");
+    let second_peer_ns = network.namespace("second_peer");
     let runtime = SharedTestRuntime::new();
 
     let first_config = test_config(
         "shared_tun_first",
         "shared_tun_network_a",
         "shared_tun_secret_a",
-        None,
+        Some(shared_ns),
         Some(&dev_name),
         "10.144.250.1/24",
     );
@@ -579,7 +585,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_second",
         "shared_tun_network_b",
         "shared_tun_secret_b",
-        None,
+        Some(shared_ns),
         Some(&dev_name),
         "10.144.251.1/24",
     );
@@ -593,7 +599,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_first_peer",
         "shared_tun_network_a",
         "shared_tun_secret_a",
-        Some(&first_peer_ns.name),
+        Some(first_peer_ns),
         None,
         "10.144.250.2/24",
     );
@@ -605,7 +611,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_second_peer",
         "shared_tun_network_b",
         "shared_tun_secret_b",
-        Some(&second_peer_ns.name),
+        Some(second_peer_ns),
         None,
         "10.144.251.2/24",
     );
@@ -648,21 +654,18 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
     )
     .await;
     wait_for_condition(
-        || async { ping_test(&first_peer_ns.name, "10.144.250.1", None).await },
+        || async { ping_test(first_peer_ns, "10.144.250.1", None).await },
         WAIT,
     )
     .await;
     wait_for_condition(
-        || async { ping_test(&second_peer_ns.name, "10.144.251.1", None).await },
+        || async { ping_test(second_peer_ns, "10.144.251.1", None).await },
         WAIT,
     )
     .await;
 
     if dhcpv6 {
-        for (instance, ns) in [
-            (&first, &first_peer_ns.name),
-            (&second, &second_peer_ns.name),
-        ] {
+        for (instance, ns) in [(&first, first_peer_ns), (&second, second_peer_ns)] {
             wait_for_condition(
                 || async { instance.get_global_ctx().get_ipv6().is_some() },
                 WAIT,
@@ -678,10 +681,10 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         }
         // Removing one DHCP member must leave the other member's addresses
         // and the physical TUN intact.
-        let original_index = tun_ifindex(ROOT_NETNS_NAME, &dev_name);
+        let original_index = tun_ifindex(shared_ns, &dev_name);
         assert!(original_index.is_some());
         drop_insts(vec![first]).await;
-        assert_eq!(tun_ifindex(ROOT_NETNS_NAME, &dev_name), original_index);
+        assert_eq!(tun_ifindex(shared_ns, &dev_name), original_index);
         let address = second
             .get_global_ctx()
             .get_ipv6()
@@ -689,7 +692,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
             .address()
             .to_string();
         wait_for_condition(
-            || async { ping_test(&second_peer_ns.name, &address, None).await },
+            || async { ping_test(second_peer_ns, &address, None).await },
             WAIT,
         )
         .await;

easytier/src/tests/three_node.rs

@@ -2075,8 +2075,13 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
                 config
                     .add_proxy_cidr("198.51.100.0/24".parse().unwrap(), None)
                     .unwrap();
+                let mut flags = config.get_flags();
+                flags.enable_exit_node = true;
+                flags.use_smoltcp = true;
+                config.set_flags(flags);
             }
             if config.get_inst_name() == "inst3" {
+                config.set_exit_nodes(vec!["10.144.144.1".parse().unwrap()]);
                 config.set_vpn_portal_config(VpnPortalConfig {
                     enabled: None,
                     wireguard_listen: "0.0.0.0:22121".parse().unwrap(),
@@ -2137,7 +2142,7 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
         dst_socket_addr,
         Key::try_from(server_public.as_slice()).unwrap(),
         Key::try_from(client_private.as_slice()).unwrap(),
-        vec!["10.144.144.0/24".to_string()],
+        vec!["10.144.144.0/24".to_string(), "10.1.1.1/32".to_string()],
         "10.144.144.4".to_owned(),
     )
     .unwrap();
@@ -2161,6 +2166,14 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
     )
     .await;
 
+    // This address is outside the mesh and advertised proxy CIDRs, on a
+    // separate bridge reachable only through the configured exit node.
+    wait_for_condition(
+        || async { ping_test(network.namespace("net_d"), "10.1.1.1", None).await },
+        Duration::from_secs(5),
+    )
+    .await;
+
     drop_insts(insts).await;
 }
 
@@ -2924,6 +2937,84 @@ pub async fn manual_reconnector(#[values(true, false)] is_foreign: bool) {
     drop_insts(vec![center_inst, inst1, inst2]).await;
 }
 
+#[rstest::rstest]
+#[tokio::test]
+#[serial_test::serial]
+async fn port_forward_survives_interface_recreation(#[values("tcp", "udp")] proto: &str) {
+    let network = prepare_linux_namespaces();
+    let namespace = network.namespace("net_a");
+    let create_tun = || {
+        run_ip_in_ns(
+            namespace,
+            &["tuntap", "add", "dev", "pf_tun", "mode", "tun"],
+        );
+        run_ip_in_ns(namespace, &["addr", "add", "192.0.2.1/24", "dev", "pf_tun"]);
+        run_ip_in_ns(namespace, &["link", "set", "pf_tun", "up"]);
+        run_ip_in_ns_output(namespace, &["-o", "link", "show", "pf_tun"])
+            .split(':')
+            .next()
+            .unwrap()
+            .to_owned()
+    };
+    let original_index = create_tun();
+    let insts = init_three_node_ex(
+        &network,
+        "udp",
+        |cfg| {
+            if cfg.get_inst_name() == "inst1" {
+                cfg.set_port_forwards(vec![PortForwardConfig {
+                    bind_addr: "192.0.2.1:23456".parse().unwrap(),
+                    dst_addr: "10.144.144.3:23456".parse().unwrap(),
+                    proto: proto.to_owned(),
+                }]);
+            }
+            cfg
+        },
+        false,
+    )
+    .await;
+
+    for recreated in [false, true] {
+        if recreated {
+            run_ip_in_ns(namespace, &["link", "del", "pf_tun"]);
+            assert_ne!(create_tun(), original_index);
+        }
+        // Keep the forwarding instance and its listener alive across deletion.
+        // Reusing the interface name and IP must not leave it bound to the old
+        // ifindex. Check an application round trip, including the return path.
+        let listener_url = format!("{proto}://0.0.0.0:23456").parse().unwrap();
+        let connector_url = format!("{proto}://192.0.2.1:23456").parse().unwrap();
+        let server_ns = NetNS::new(Some(network.namespace("net_c").into()));
+        let client_ns = NetNS::new(Some(namespace.into()));
+        let payload = b"port-forward interface recreation".to_vec();
+        let timeout = Duration::from_secs(5);
+        let result = if proto == "tcp" {
+            _tunnel_pingpong_netns_with_timeout(
+                core_tcp_listener(listener_url),
+                core_tcp_dialer(connector_url),
+                server_ns,
+                client_ns,
+                payload,
+                timeout,
+            )
+            .await
+        } else {
+            _tunnel_pingpong_netns_with_timeout(
+                core_udp_listener(listener_url),
+                core_udp_dialer(connector_url),
+                server_ns,
+                client_ns,
+                payload,
+                timeout,
+            )
+            .await
+        };
+        assert!(result.is_ok(), "{proto}, recreated={recreated}: {result:?}");
+    }
+
+    drop_insts(insts).await;
+}
+
 #[rstest::rstest]
 #[tokio::test]
 #[serial_test::serial]
@@ -5400,14 +5491,45 @@ pub async fn dhcpv6_three_node_connectivity(#[case] dhcp_ipv4: bool, #[case] sta
     for address in &addresses {
         assert_eq!(address.network(), "fd42:1234::/64".parse().unwrap());
     }
+    // DHCP can temporarily assign the same IPv4 address to multiple peers.
+    // Wait for conflict resolution and route propagation in both directions
+    // before opening connections; IPv6 readiness alone does not imply this.
     wait_for_condition(
         || async {
-            insts[0]
-                .get_core_instance()
-                .route_snapshots()
-                .await
+            let ipv4_addresses = insts
                 .iter()
-                .all(|route| route.ipv6_addr.is_some())
+                .map(|inst| inst.get_global_ctx().get_ipv4())
+                .collect::<Vec<_>>();
+            if (dhcp_ipv4 || static_ipv4)
+                && ipv4_addresses
+                    .iter()
+                    .flatten()
+                    .map(|address| address.address())
+                    .collect::<std::collections::HashSet<_>>()
+                    .len()
+                    != insts.len()
+            {
+                return false;
+            }
+            for inst in &insts {
+                let routes = inst.get_core_instance().route_snapshots().await;
+                if routes.len() != insts.len() - 1 {
+                    return false;
+                }
+                for (index, peer) in insts.iter().enumerate() {
+                    if peer.peer_id() == inst.peer_id() {
+                        continue;
+                    }
+                    if !routes.iter().any(|route| {
+                        route.peer_id == peer.peer_id()
+                            && route.ipv6_addr == Some(addresses[index].into())
+                            && route.ipv4_addr == ipv4_addresses[index].map(Into::into)
+                    }) {
+                        return false;
+                    }
+                }
+            }
+            true
         },
         Duration::from_secs(15),
     )

easytier/src/tunnel/wireguard.rs

@@ -843,13 +843,32 @@ pub mod tests {
     }
 
     #[tokio::test]
-    async fn bind_same_port() {
-        let server_cfg = test_wg_config();
-        let mut listener = WgTunnelListener::new("wg://[::1]:31015".parse().unwrap(), server_cfg);
-        let server_cfg = test_wg_config();
-        let mut listener2 = WgTunnelListener::new("wg://[::1]:31015".parse().unwrap(), server_cfg);
+    async fn bind_ipv4_and_ipv6_same_port() {
+        let mut listener = WgTunnelListener::new("wg://[::]:0".parse().unwrap(), test_wg_config());
         listener.listen().await.unwrap();
+        let port = listener.local_url().port().unwrap();
+
+        let mut listener2 = WgTunnelListener::new(
+            format!("wg://0.0.0.0:{port}").parse().unwrap(),
+            test_wg_config(),
+        );
         listener2.listen().await.unwrap();
+        assert_eq!(listener2.local_url().port(), Some(port));
+    }
+
+    #[tokio::test]
+    async fn bind_same_port_is_rejected() {
+        let server_cfg = test_wg_config();
+        let mut listener = WgTunnelListener::new("wg://[::1]:0".parse().unwrap(), server_cfg);
+        listener.listen().await.unwrap();
+
+        let mut listener2 = WgTunnelListener::new(listener.local_url(), test_wg_config());
+        let error = listener2.listen().await.unwrap_err();
+        assert!(error.chain().any(|cause| {
+            cause
+                .downcast_ref::<std::io::Error>()
+                .is_some_and(|error| error.kind() == std::io::ErrorKind::AddrInUse)
+        }));
     }
 
     #[tokio::test]