fix(tun): preserve shared NIC ownership and failure invariants

Keep mobile address and route claims in dispatcher tables without
applying desktop interface configuration to synthetic TUN names.

Limit the Magic DNS overlap exemption to route claims and remove the
shared route retry that bypassed claim accounting after invalidation.

Invalidate and stop dispatchers on failed mobile TUN replacement or I/O.
Bound TUN writes so stalled devices release pending source updates and
member cleanup instead of retaining an unusable shared dispatcher.

Treat missing Darwin and Windows addresses and routes as successful
deletions while preserving permission and other configuration errors.

Add regressions for claim conflicts, mobile configuration, failed TUN
replacement, stalled writes, and platform deletion error handling.

docs/core-architecture.md

@@ -430,6 +430,9 @@ the platform VPN owns its addresses, routes, and MTU.
 If applying native claims fails, the shared NIC is invalidated and its
 dispatcher is stopped so every member can recreate the device. Restoring only
 the recorded claims would leave partially applied OS configuration in use.
+TUN replacement and I/O failures also invalidate the shared NIC. TUN writes
+have bounded waits so a stalled device cannot indefinitely block claim updates
+or member cleanup.
 
 Desktop and mobile shared TUNs use the same lazily initialized native Tokio
 runtime, independent of every member's executor. It creates the device's async

easytier/src/common/ifcfg/darwin.rs

@@ -1,6 +1,8 @@
 use std::{collections::BTreeSet, net::Ipv4Addr};
 
-use super::{Error, IfConfiguerTrait, cidr_to_subnet_mask, run_shell_cmd};
+use super::{
+    Error, IfConfiguerTrait, cidr_to_subnet_mask, ignore_removed_shell_cmd_not_found, run_shell_cmd,
+};
 use async_trait::async_trait;
 use cidr::{Ipv4Inet, Ipv6Inet};
 use tokio::sync::Mutex;
@@ -33,6 +35,10 @@ impl MacIfConfiger {
     }
 }
 
+async fn run_delete_shell_cmd(cmd: &str) -> Result<(), Error> {
+    ignore_removed_shell_cmd_not_found(run_shell_cmd(cmd).await)
+}
+
 #[async_trait]
 impl IfConfiguerTrait for MacIfConfiger {
     async fn add_ipv4_route(
@@ -77,7 +83,7 @@ impl IfConfiguerTrait for MacIfConfiger {
         address: Ipv4Addr,
         cidr_prefix: u8,
     ) -> Result<(), Error> {
-        run_shell_cmd(
+        run_delete_shell_cmd(
             format!(
                 "route -n delete {} -netmask {} -interface {}",
                 address,
@@ -114,11 +120,13 @@ impl IfConfiguerTrait for MacIfConfiger {
     async fn remove_ip(&self, name: &str, ip: Option<Ipv4Inet>) -> Result<(), Error> {
         let mut configured_ipv4 = self.configured_ipv4.lock().await;
         if let Some(ip) = ip {
-            run_shell_cmd(format!("ifconfig {} inet {} delete", name, ip.address()).as_str())
-                .await?;
+            run_delete_shell_cmd(
+                format!("ifconfig {} inet {} delete", name, ip.address()).as_str(),
+            )
+            .await?;
             configured_ipv4.remove(&ip);
         } else {
-            run_shell_cmd(format!("ifconfig {} inet delete", name).as_str()).await?;
+            run_delete_shell_cmd(format!("ifconfig {} inet delete", name).as_str()).await?;
             configured_ipv4.clear();
         }
         Ok(())
@@ -140,7 +148,10 @@ impl IfConfiguerTrait for MacIfConfiger {
 
     async fn remove_ipv6(&self, name: &str, ip: Option<Ipv6Inet>) -> Result<(), Error> {
         if let Some(ip) = ip {
-            run_shell_cmd(format!("ifconfig {} inet6 {} delete", name, ip.address()).as_str()).await
+            run_delete_shell_cmd(
+                format!("ifconfig {} inet6 {} delete", name, ip.address()).as_str(),
+            )
+            .await
         } else {
             // Remove all IPv6 addresses is more complex on macOS, just succeed
             Ok(())
@@ -174,7 +185,7 @@ impl IfConfiguerTrait for MacIfConfiger {
         address: std::net::Ipv6Addr,
         cidr_prefix: u8,
     ) -> Result<(), Error> {
-        run_shell_cmd(
+        run_delete_shell_cmd(
             format!(
                 "route -n delete -inet6 {}/{} -interface {}",
                 address, cidr_prefix, name

easytier/src/common/ifcfg/mod.rs

@@ -162,7 +162,12 @@ async fn run_shell_cmd(cmd: &str) -> Result<(), Error> {
 
     #[cfg(not(target_os = "windows"))]
     {
-        cmd_out = Command::new("sh").arg("-c").arg(cmd).output().await?;
+        cmd_out = Command::new("sh")
+            .env("LC_ALL", "C")
+            .arg("-c")
+            .arg(cmd)
+            .output()
+            .await?;
         stdout = String::from_utf8_lossy(cmd_out.stdout.as_slice()).to_string();
         stderr = String::from_utf8_lossy(cmd_out.stderr.as_slice()).to_string();
     };
@@ -177,6 +182,33 @@ async fn run_shell_cmd(cmd: &str) -> Result<(), Error> {
     Ok(())
 }
 
+#[cfg(any(
+    all(target_os = "macos", not(feature = "macos-ne")),
+    target_os = "freebsd",
+    test
+))]
+fn ignore_removed_shell_cmd_not_found(result: Result<(), Error>) -> Result<(), Error> {
+    match result {
+        Err(Error::ShellCommandError(output))
+            if !output.trim().is_empty()
+                && output
+                    .lines()
+                    .filter(|line| !line.trim().is_empty())
+                    .all(|line| {
+                        (line.starts_with("ifconfig: ")
+                            && line.ends_with(": Can't assign requested address"))
+                            || (line.starts_with("delete ") && line.ends_with(": not in table"))
+                            || line == "route: route has not been found"
+                            || line == "route: writing to routing socket: not in table"
+                            || line == "route: writing to routing socket: No such process"
+                    }) =>
+        {
+            Ok(())
+        }
+        other => other,
+    }
+}
+
 #[derive(Default)]
 pub struct DummyIfConfiger {}
 #[async_trait]
@@ -242,3 +274,54 @@ pub(crate) fn list_ipv6_ndp_proxy(
 ) -> Result<std::collections::BTreeSet<Ipv6Addr>, Error> {
     netlink::NetlinkIfConfiger::list_ipv6_ndp_proxy(name)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{Error, ignore_removed_shell_cmd_not_found};
+
+    #[test]
+    fn removed_shell_cmd_ignores_missing_addresses_and_routes() {
+        for output in [
+            "ifconfig: ioctl (SIOCDIFADDR): Can't assign requested address\n",
+            "ifconfig: ioctl (SIOCDIFADDR_IN6): Can't assign requested address\n",
+            "delete net 10.0.0.0: gateway tun0: not in table\n",
+            "delete net 10.0.0.0: gateway tun0 fib 0: not in table\nroute: route has not been found\n",
+            "delete net 10.0.0.0: gateway tun0 fib 0: not in table\nroute: writing to routing socket: No such process\n",
+            "delete net 10.0.0.0: gateway tun0: not in table\nroute: writing to routing socket: not in table\n",
+        ] {
+            assert!(
+                ignore_removed_shell_cmd_not_found(Err(Error::ShellCommandError(
+                    output.to_owned()
+                )))
+                .is_ok()
+            );
+        }
+    }
+
+    #[test]
+    fn removed_shell_cmd_preserves_other_errors() {
+        for output in [
+            "ifconfig: ioctl (SIOCDIFADDR): Operation not permitted\n",
+            "route: writing to routing socket: Permission denied\n",
+            "route: route has not been found: Permission denied\n",
+            "route: route has not been found: No such process\n",
+            "delete net 10.0.0.0: gateway tun0: not in table\nroute: Permission denied\n",
+            "delete net 10.0.0.0: gateway tun0: not in table\nroute: writing to routing socket: No such process\nroute: Permission denied\n",
+            "unexpected command: not in table\n",
+            "",
+        ] {
+            assert!(matches!(
+                ignore_removed_shell_cmd_not_found(Err(Error::ShellCommandError(output.to_owned()))),
+                Err(Error::ShellCommandError(actual)) if actual == output
+            ));
+        }
+
+        assert!(matches!(
+            ignore_removed_shell_cmd_not_found(Err(std::io::Error::from(
+                std::io::ErrorKind::PermissionDenied
+            )
+            .into())),
+            Err(Error::IOError(_))
+        ));
+    }
+}

easytier/src/common/ifcfg/windows.rs

@@ -9,7 +9,7 @@ use std::{
 };
 use windows::Win32::NetworkManagement::IpHelper::INTERNAL_IF_OPER_STATUS;
 use windows::Win32::{
-    Foundation::NO_ERROR,
+    Foundation::{ERROR_NOT_FOUND, NO_ERROR},
     NetworkManagement::IpHelper::{GetIfEntry, MIB_IFROW, SetIfEntry},
     System::Diagnostics::Debug::{
         FORMAT_MESSAGE_FROM_SYSTEM, FORMAT_MESSAGE_IGNORE_INSERTS, FormatMessageW,
@@ -25,6 +25,13 @@ use super::{Error, IfConfiguerTrait};
 #[derive(Default)]
 pub struct WindowsIfConfiger {}
 
+fn ignore_removed_not_found(result: Result<(), u32>) -> Result<(), u32> {
+    match result {
+        Err(error) if error == ERROR_NOT_FOUND.0 => Ok(()),
+        other => other,
+    }
+}
+
 fn format_win_error(error: u32) -> String {
     // use FormatMessageW to get the error message
     let mut buffer = vec![0; 1024];
@@ -75,7 +82,7 @@ impl WindowsIfConfiger {
             anyhow::anyhow!("Failed to get interface luid: {}", format_win_error(e))
         })?;
         if let Some(addr) = addr {
-            luid.delete_ipv4_address(&addr).map_err(|e| {
+            ignore_removed_not_found(luid.delete_ipv4_address(&addr)).map_err(|e| {
                 anyhow::anyhow!("Failed to delete IPv4 address: {}", format_win_error(e))
             })?;
         } else {
@@ -168,7 +175,7 @@ impl WindowsIfConfiger {
             anyhow::anyhow!("Failed to get interface luid: {}", format_win_error(e))
         })?;
         if let Some(addr) = addr {
-            luid.delete_ipv6_address(&addr).map_err(|e| {
+            ignore_removed_not_found(luid.delete_ipv6_address(&addr)).map_err(|e| {
                 anyhow::anyhow!("Failed to delete IPv6 address: {}", format_win_error(e))
             })?;
         } else {
@@ -217,10 +224,10 @@ impl IfConfiguerTrait for WindowsIfConfiger {
         let luid = InterfaceLuid::luid_from_index(if_index).map_err(|e| {
             anyhow::anyhow!("Failed to get interface luid: {}", format_win_error(e))
         })?;
-        luid.delete_route_ipv4(
+        ignore_removed_not_found(luid.delete_route_ipv4(
             &Ipv4Inet::new(address, cidr_prefix).unwrap(),
             &Ipv4Addr::UNSPECIFIED,
-        )
+        ))
         .map_err(|e| anyhow::anyhow!("Failed to delete route: {}", format_win_error(e)))?;
         Ok(())
     }
@@ -307,10 +314,10 @@ impl IfConfiguerTrait for WindowsIfConfiger {
             anyhow::anyhow!("Failed to get interface luid: {}", format_win_error(e))
         })?;
 
-        luid.delete_route_ipv6(
+        ignore_removed_not_found(luid.delete_route_ipv6(
             &Ipv6Inet::new(address, cidr_prefix).unwrap(),
             &Ipv6Addr::UNSPECIFIED,
-        )
+        ))
         .map_err(|e| anyhow::anyhow!("Failed to delete route: {}", format_win_error(e)))?;
         Ok(())
     }
@@ -531,3 +538,24 @@ impl RegistryManager {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{ERROR_NOT_FOUND, ignore_removed_not_found};
+    use windows::Win32::Foundation::{
+        ERROR_ACCESS_DENIED, ERROR_FILE_NOT_FOUND, ERROR_INVALID_PARAMETER,
+    };
+
+    #[test]
+    fn removed_windows_ifcfg_ignores_only_not_found() {
+        assert_eq!(ignore_removed_not_found(Ok(())), Ok(()));
+        assert_eq!(ignore_removed_not_found(Err(ERROR_NOT_FOUND.0)), Ok(()));
+        for error in [
+            ERROR_ACCESS_DENIED.0,
+            ERROR_INVALID_PARAMETER.0,
+            ERROR_FILE_NOT_FOUND.0,
+        ] {
+            assert_eq!(ignore_removed_not_found(Err(error)), Err(error));
+        }
+    }
+}

easytier/src/instance/dns_server/runner.rs

@@ -5,15 +5,7 @@ use std::{net::Ipv4Addr, sync::Arc, time::Duration};
 
 use easytier_core::instance::CorePacketPlane;
 
-use crate::{
-    common::{
-        error::Error as EtError,
-        global_ctx::ArcGlobalCtx,
-        ifcfg::{IfConfiger, IfConfiguerTrait},
-        netns::NetNS,
-    },
-    instance::virtual_nic::NicBackend,
-};
+use crate::{common::global_ctx::ArcGlobalCtx, instance::virtual_nic::NicBackend};
 
 use super::{client_instance::MagicDnsClientInstance, server_instance::MagicDnsServerInstance};
 
@@ -30,8 +22,6 @@ pub struct DnsRunner {
 }
 
 struct MagicDnsFakeIpRouteClaim {
-    tun_dev: Option<String>,
-    net_ns: NetNS,
     fake_ip: Ipv4Addr,
     route_backend: NicBackend,
 }
@@ -44,26 +34,10 @@ impl MagicDnsFakeIpRouteClaim {
             None
         };
 
-        match self
-            .route_backend
+        self.route_backend
             .add_route_with_cost(self.fake_ip, 32, cost)
-            .await
-        {
-            Err(EtError::IOError(err))
-                if err.kind() == std::io::ErrorKind::AlreadyExists && self.tun_dev.is_some() =>
-            {
-                let ifcfg = IfConfiger::default();
-                let _guard = self.net_ns.guard();
-                ifcfg
-                    .remove_ipv4_route(self.tun_dev.as_deref().unwrap(), self.fake_ip, 32)
-                    .await?;
-                self.route_backend
-                    .add_route_with_cost(self.fake_ip, 32, cost)
-                    .await?;
-                Ok(())
-            }
-            result => result.map_err(Into::into),
-        }
+            .await?;
+        Ok(())
     }
 
     async fn remove(&self) {
@@ -119,8 +93,6 @@ impl DnsRunner {
         }
 
         Some(MagicDnsFakeIpRouteClaim {
-            tun_dev: self.tun_dev.clone(),
-            net_ns: self.global_ctx.net_ns.clone(),
             fake_ip: self.fake_ip,
             route_backend: self.shared_route_backend.clone()?,
         })

easytier/src/instance/shared_virtual_nic.rs

@@ -705,7 +705,10 @@ impl SharedVirtualNic {
 
         if let Some(dispatcher) = &self.dispatcher {
             if replace_tun_fd {
-                dispatcher.update_mobile_tun_fd(tun_fd).await?;
+                if let Err(err) = dispatcher.update_mobile_tun_fd(tun_fd).await {
+                    self.invalidate_and_shutdown_dispatcher().await;
+                    return Err(err);
+                }
             }
             return Ok(());
         }

easytier/src/instance/shared_virtual_nic/dispatcher.rs

@@ -14,13 +14,13 @@ use smoltcp::wire::{
     Icmpv4Message, Icmpv4Packet, Icmpv6Message, Icmpv6Packet, IpAddress, IpProtocol, Ipv4Packet,
     Ipv6Packet, TcpPacket, UdpPacket,
 };
-#[cfg(mobile)]
+#[cfg(any(mobile, all(test, unix)))]
 use tokio::sync::Mutex;
 use tokio::sync::{Notify, mpsc, oneshot};
 use tokio_util::task::AbortOnDropHandle;
 
 use crate::common::error::Error;
-#[cfg(mobile)]
+#[cfg(any(mobile, all(test, unix)))]
 use crate::instance::virtual_nic::VirtualNic;
 use easytier_core::{
     packet::ZCPacket,
@@ -37,13 +37,14 @@ const FLOW_OWNER_LIMIT: usize = 4096;
 // Match the core UDP proxy's idle lifetime; reclaim lazily under table pressure.
 const FLOW_IDLE_TIMEOUT: Duration = Duration::from_secs(180);
 const DISPATCHER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
+const DISPATCHER_TUN_WRITE_TIMEOUT: Duration = Duration::from_secs(1);
 
 struct SharedVirtualNicMemberPacket {
     member_id: SharedVirtualNicMemberId,
     packet: ZCPacket,
 }
 
-#[cfg(mobile)]
+#[cfg(any(mobile, all(test, unix)))]
 struct SharedVirtualNicMobileTunUpdate {
     tun_fd: std::os::fd::RawFd,
     ack: oneshot::Sender<Result<(), Error>>,
@@ -781,6 +782,14 @@ fn acknowledge_dispatcher_shutdown(ack: Option<oneshot::Sender<()>>) {
     }
 }
 
+async fn send_packet_to_tun(
+    tun_sink: &mut Pin<Box<dyn ZCPacketSink>>,
+    packet: ZCPacket,
+) -> Result<(), Error> {
+    tokio::time::timeout(DISPATCHER_TUN_WRITE_TIMEOUT, tun_sink.send(packet)).await??;
+    Ok(())
+}
+
 struct SharedVirtualNicDispatcherTask {
     tun_stream: Pin<Box<dyn ZCPacketStream>>,
     tun_sink: Pin<Box<dyn ZCPacketSink>>,
@@ -849,7 +858,7 @@ impl SharedVirtualNicDispatcherTask {
         else {
             return true;
         };
-        if let Err(err) = self.tun_sink.send(packet).await {
+        if let Err(err) = send_packet_to_tun(&mut self.tun_sink, packet).await {
             tracing::error!(?err, "shared virtual nic write to tun failed");
             return false;
         }
@@ -857,7 +866,7 @@ impl SharedVirtualNicDispatcherTask {
     }
 }
 
-#[cfg(mobile)]
+#[cfg(any(mobile, all(test, unix)))]
 struct SharedVirtualNicMobileDispatcherTask {
     nic: Arc<Mutex<VirtualNic>>,
     tun_stream: Option<Pin<Box<dyn ZCPacketStream>>>,
@@ -871,8 +880,9 @@ struct SharedVirtualNicMobileDispatcherTask {
     state: SharedVirtualNicDispatcherState,
 }
 
-#[cfg(mobile)]
+#[cfg(any(mobile, all(test, unix)))]
 impl SharedVirtualNicMobileDispatcherTask {
+    #[cfg(mobile)]
     async fn run(mut self, first_open_sender: oneshot::Sender<Result<(), Error>>) {
         match self.open_tun().await {
             Ok(()) => {
@@ -888,6 +898,10 @@ impl SharedVirtualNicMobileDispatcherTask {
             }
         }
 
+        self.run_opened_tun().await;
+    }
+
+    async fn run_opened_tun(mut self) {
         loop {
             tokio::select! {
                 control = self.control_receiver.recv() => {
@@ -900,7 +914,10 @@ impl SharedVirtualNicMobileDispatcherTask {
                         self.cleanup(true);
                         return;
                     };
-                    self.forward_member_packet_to_tun(member_packet).await;
+                    if !self.forward_member_packet_to_tun(member_packet).await {
+                        self.cleanup(true);
+                        return;
+                    }
                 }
                 packet = async {
                     match self.tun_stream.as_mut() {
@@ -910,14 +927,15 @@ impl SharedVirtualNicMobileDispatcherTask {
                 } => {
                     let Some(packet) = packet else {
                         tracing::error!("shared virtual nic mobile tun stream closed");
-                        self.drop_tun();
-                        continue;
+                        self.cleanup(true);
+                        return;
                     };
                     match packet {
                         Ok(packet) => self.state.forward_tun_packet_to_member(packet).await,
                         Err(err) => {
                             tracing::error!(?err, "shared virtual nic read from mobile tun failed");
-                            self.drop_tun();
+                            self.cleanup(true);
+                            return;
                         }
                     }
                 }
@@ -926,7 +944,9 @@ impl SharedVirtualNicMobileDispatcherTask {
                         self.cleanup(true);
                         return;
                     };
-                    self.apply_tun_update(update).await;
+                    if !self.apply_tun_update(update).await {
+                        return;
+                    }
                 }
             }
         }
@@ -942,39 +962,42 @@ impl SharedVirtualNicMobileDispatcherTask {
         Ok(())
     }
 
-    async fn apply_tun_update(&mut self, update: SharedVirtualNicMobileTunUpdate) {
+    async fn apply_tun_update(&mut self, update: SharedVirtualNicMobileTunUpdate) -> bool {
         self.drop_tun();
         self.tun_fd = update.tun_fd;
         match self.open_tun().await {
             Ok(()) => {
                 let _ = update.ack.send(Ok(()));
+                true
             }
             Err(err) => {
                 tracing::error!(fd = self.tun_fd, ?err, "failed to replace mobile tun");
+                self.cleanup(true);
                 let _ = update.ack.send(Err(err));
+                false
             }
         }
     }
 
-    async fn forward_member_packet_to_tun(&mut self, member_packet: SharedVirtualNicMemberPacket) {
+    async fn forward_member_packet_to_tun(
+        &mut self,
+        member_packet: SharedVirtualNicMemberPacket,
+    ) -> bool {
         let Some(packet) = self
             .state
             .prepare_member_packet_to_tun(member_packet.member_id, member_packet.packet)
         else {
-            return;
+            return true;
         };
         let Some(tun_sink) = self.tun_sink.as_mut() else {
-            tracing::trace!(
-                member_id = ?member_packet.member_id,
-                "shared virtual nic dropped member packet without mobile tun"
-            );
-            return;
+            return false;
         };
 
-        if let Err(err) = tun_sink.send(packet).await {
+        if let Err(err) = send_packet_to_tun(tun_sink, packet).await {
             tracing::error!(?err, "shared virtual nic write to mobile tun failed");
-            self.drop_tun();
+            return false;
         }
+        true
     }
 
     fn handle_control(&mut self, control: Option<SharedVirtualNicControl>) -> bool {
@@ -1981,6 +2004,240 @@ mod tests {
         }
     }
 
+    #[cfg(unix)]
+    fn start_mobile_dispatcher_for_test(
+        tunnel: Box<dyn Tunnel>,
+        member_tunnel_table: SharedVirtualNicMemberTunnelTable,
+        valid: Arc<AtomicBool>,
+    ) -> (
+        SharedVirtualNicDispatcher,
+        mpsc::UnboundedSender<SharedVirtualNicMobileTunUpdate>,
+    ) {
+        use crate::{common::netns::NetNS, instance::virtual_nic::VirtualNicConfig};
+
+        let (tun_stream, tun_sink) = tunnel.split();
+        let (to_tun_sender, to_tun_receiver) = mpsc::channel(MEMBER_TUNNEL_BUFFER_SIZE);
+        let (control_sender, control_receiver) = mpsc::unbounded_channel();
+        let (tun_update_sender, tun_update_receiver) = mpsc::unbounded_channel();
+        member_tunnel_table.attach_dispatcher(to_tun_sender, control_sender.clone());
+        let task = SharedVirtualNicMobileDispatcherTask {
+            nic: Arc::new(Mutex::new(VirtualNic::new(VirtualNicConfig::new(
+                String::new(),
+                1300,
+                NetNS::new(None),
+            )))),
+            tun_stream: Some(tun_stream),
+            tun_sink: Some(tun_sink),
+            tun_fd: -1,
+            tun_update_receiver,
+            to_tun_receiver,
+            control_receiver,
+            member_tunnel_table,
+            valid,
+            state: SharedVirtualNicDispatcherState {
+                translate_unknown_sources: true,
+                ..Default::default()
+            },
+        };
+        let dispatcher = SharedVirtualNicDispatcher {
+            _task: AbortOnDropHandle::new(tokio::spawn(task.run_opened_tun())),
+            control_sender,
+            #[cfg(mobile)]
+            mobile_tun_update_sender: Some(tun_update_sender.clone()),
+        };
+        (dispatcher, tun_update_sender)
+    }
+
+    async fn register_dispatcher_test_members(
+        dispatcher: &SharedVirtualNicDispatcher,
+        member_tunnel_table: &SharedVirtualNicMemberTunnelTable,
+    ) -> (Vec<Box<dyn Tunnel>>, Vec<Arc<Notify>>) {
+        let mut tunnels = Vec::new();
+        let mut close_notifiers = Vec::new();
+        for id in 1..=2 {
+            let member_id = uuid::Uuid::from_u128(id);
+            let close_notifier = Arc::new(Notify::new());
+            let (member_tunnel, shared_tunnel) = create_ring_tunnel_pair();
+            member_tunnel_table
+                .register(
+                    member_id,
+                    uuid::Uuid::new_v4(),
+                    shared_tunnel,
+                    close_notifier.clone(),
+                )
+                .unwrap();
+            dispatcher
+                .update_sources(member_id, &SharedIfConfigClaims::default())
+                .await
+                .unwrap();
+            tunnels.push(member_tunnel);
+            close_notifiers.push(close_notifier);
+        }
+        (tunnels, close_notifiers)
+    }
+
+    fn stalled_tun_for_test() -> (Box<dyn Tunnel>, Arc<Notify>) {
+        let writing = Arc::new(Notify::new());
+        let sink_writing = writing.clone();
+        let tun_sink = futures::sink::unfold((), move |(), _packet: ZCPacket| {
+            let writing = sink_writing.clone();
+            async move {
+                writing.notify_one();
+                std::future::pending::<Result<(), TunnelError>>().await
+            }
+        });
+        let tun_stream = futures::stream::pending::<Result<ZCPacket, TunnelError>>();
+        (
+            Box::new(TunnelWrapper::new(tun_stream, tun_sink, None)),
+            writing,
+        )
+    }
+
+    async fn check_dispatcher_tun_write_timeout(
+        dispatcher: SharedVirtualNicDispatcher,
+        member_tunnel_table: SharedVirtualNicMemberTunnelTable,
+        valid: Arc<AtomicBool>,
+        writing: Arc<Notify>,
+    ) {
+        let (members, close_notifiers) =
+            register_dispatcher_test_members(&dispatcher, &member_tunnel_table).await;
+        let (_, mut member_sink) = members[0].split();
+        member_sink
+            .send(ipv4_udp_packet(
+                Ipv4Addr::new(10, 231, 1, 1),
+                Ipv4Addr::new(203, 0, 113, 1),
+            ))
+            .await
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(1), writing.notified())
+            .await
+            .unwrap();
+
+        let result = tokio::time::timeout(
+            DISPATCHER_TUN_WRITE_TIMEOUT + Duration::from_secs(1),
+            dispatcher.remove_sources(uuid::Uuid::from_u128(1)),
+        )
+        .await
+        .unwrap();
+        assert!(result.is_err());
+        for close_notifier in close_notifiers {
+            tokio::time::timeout(Duration::from_secs(1), close_notifier.notified())
+                .await
+                .unwrap();
+        }
+        assert!(!valid.load(Ordering::Acquire));
+        assert!(member_tunnel_table.dispatcher_channels().is_none());
+        tokio::time::timeout(Duration::from_secs(1), dispatcher._task)
+            .await
+            .unwrap()
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn dispatcher_tun_write_timeout_closes_members_and_source_updates() {
+        let (tunnel, writing) = stalled_tun_for_test();
+        let member_tunnel_table = SharedVirtualNicMemberTunnelTable::default();
+        let valid = Arc::new(AtomicBool::new(true));
+        let dispatcher =
+            SharedVirtualNicDispatcher::start(tunnel, member_tunnel_table.clone(), valid.clone());
+
+        check_dispatcher_tun_write_timeout(dispatcher, member_tunnel_table, valid, writing).await;
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn mobile_dispatcher_tun_write_timeout_closes_members_and_source_updates() {
+        let (tunnel, writing) = stalled_tun_for_test();
+        let member_tunnel_table = SharedVirtualNicMemberTunnelTable::default();
+        let valid = Arc::new(AtomicBool::new(true));
+        let (dispatcher, _updates) =
+            start_mobile_dispatcher_for_test(tunnel, member_tunnel_table.clone(), valid.clone());
+
+        check_dispatcher_tun_write_timeout(dispatcher, member_tunnel_table, valid, writing).await;
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn mobile_dispatcher_invalidates_shared_nic_when_tun_replacement_fails() {
+        let (_tun, tunnel) = create_ring_tunnel_pair();
+        let member_tunnel_table = SharedVirtualNicMemberTunnelTable::default();
+        let valid = Arc::new(AtomicBool::new(true));
+        let (dispatcher, updates) =
+            start_mobile_dispatcher_for_test(tunnel, member_tunnel_table.clone(), valid.clone());
+        let (_members, close_notifiers) =
+            register_dispatcher_test_members(&dispatcher, &member_tunnel_table).await;
+        let (ack, receiver) = oneshot::channel();
+
+        updates
+            .send(SharedVirtualNicMobileTunUpdate { tun_fd: -1, ack })
+            .unwrap();
+
+        assert!(
+            tokio::time::timeout(Duration::from_secs(1), await_mobile_open(receiver))
+                .await
+                .unwrap()
+                .is_err()
+        );
+        for close_notifier in close_notifiers {
+            tokio::time::timeout(Duration::from_secs(1), close_notifier.notified())
+                .await
+                .unwrap();
+        }
+        assert!(!valid.load(Ordering::Acquire));
+        assert!(member_tunnel_table.dispatcher_channels().is_none());
+        assert!(
+            dispatcher
+                .remove_sources(uuid::Uuid::from_u128(1))
+                .await
+                .is_err()
+        );
+        tokio::time::timeout(Duration::from_secs(1), dispatcher._task)
+            .await
+            .unwrap()
+            .unwrap();
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn mobile_dispatcher_invalidates_shared_nic_when_tun_read_closes() {
+        for read_error in [false, true] {
+            let (tun_tx, tun_rx) = mpsc::unbounded_channel();
+            let tun_stream = futures::stream::unfold(tun_rx, |mut receiver| async move {
+                receiver.recv().await.map(|packet| (packet, receiver))
+            });
+            let tun_sink = futures::sink::unfold((), |(), _packet: ZCPacket| async {
+                Ok::<(), TunnelError>(())
+            });
+            let tunnel = Box::new(TunnelWrapper::new(tun_stream, tun_sink, None));
+            let member_tunnel_table = SharedVirtualNicMemberTunnelTable::default();
+            let valid = Arc::new(AtomicBool::new(true));
+            let (dispatcher, _updates) = start_mobile_dispatcher_for_test(
+                tunnel,
+                member_tunnel_table.clone(),
+                valid.clone(),
+            );
+            let (_members, close_notifiers) =
+                register_dispatcher_test_members(&dispatcher, &member_tunnel_table).await;
+
+            if read_error {
+                tun_tx.send(Err(TunnelError::Shutdown)).unwrap();
+            }
+            drop(tun_tx);
+
+            for close_notifier in close_notifiers {
+                tokio::time::timeout(Duration::from_secs(1), close_notifier.notified())
+                    .await
+                    .unwrap();
+            }
+            assert!(!valid.load(Ordering::Acquire));
+            assert!(member_tunnel_table.dispatcher_channels().is_none());
+            tokio::time::timeout(Duration::from_secs(1), dispatcher._task)
+                .await
+                .unwrap()
+                .unwrap();
+        }
+    }
+
     #[test]
     fn source_table_selects_ipv6_source_owner() {
         let first = uuid::Uuid::from_u128(1);

easytier/src/instance/virtual_nic.rs

@@ -600,12 +600,12 @@ impl VirtualNic {
         Ok(tun::create(&config)?)
     }
 
-    #[cfg(mobile)]
+    #[cfg(any(mobile, all(test, unix)))]
     pub fn set_mobile_tun_fd_name(&mut self, tun_fd: std::os::fd::RawFd) {
         self.ifname = Some(format!("tunfd_{}", tun_fd));
     }
 
-    #[cfg(mobile)]
+    #[cfg(any(mobile, all(test, unix)))]
     pub async fn create_dev_for_mobile(
         &mut self,
         tun_fd: std::os::fd::RawFd,