fix(entry): prevent stale peer eviction and TUN EBUSY on reconnect

Three interconnected TUN lifecycle bugs blocked reliable reconnects:

1. **Stale peer registration race:** Accept loop registers a new connection
   before the old task's unregister fires, evicting the new peer. Fix:
   monotonic connection_id on PeerInfo; unregister_if_current only removes
   when the ID matches.

2. **TUN EBUSY on reconnect:** SYN probe tasks held Arc<AsyncFd<Device>>
   clones after ConnectionManager exited. Fix: probe tasks run in a JoinSet
   that is drained before returning; run_connection_loop aborts+joins the
   manager task; SessionManager.get_or_create preemptively deletes stale
   TUN interfaces before reuse.

3. **TUN deletion before route cleanup:** delete_tun ran inside
   ConnectionParams::run() before the outer task removed OS routes. Fix:
   delete_tun sequenced after run_connection_loop abort+join (fd released);
   TunDropGuard added for panic safety.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: maxholman

TODO.md

@@ -11,6 +11,22 @@
 - [ ] Reduce global lock contention in entry-stack
 - [x] ~~Broadcast → mpsc migration on data path~~ — done.
 
+## Relay
+
+- [ ] **Bidi stream bridging** — relay currently bridges uni streams only.
+      Bidi streams (used by SYN probes and TCP data sessions) go point-to-point
+      and are invisible to the relay. Entry's `probe_tcp_target` and
+      `run_tcp_session` both open bidi streams that hang when a relay is in
+      the path. Fix: relay accepts bidi from one side, opens matching bidi to
+      the other side, splices with `copy_bidirectional`. Unblocks SYN probes
+      and TCP data flow through relay chains without any entry/exit changes.
+- [ ] **Topology visibility** — entry has no visibility into peers behind a
+      relay. `wallhack peers` shows the relay but not the exit nodes connected
+      to it. Relay should forward downstream peer identity (name, capabilities,
+      routes) upstream via a peer announcement control message or augmented
+      handshake. Needed for topology observability, debugging, and multi-hop
+      route selection.
+
 ## Transports
 
 - [ ] DNS over tunnel — intercept DNS queries at the TUN interface on the exit
@@ -82,11 +98,8 @@
       impossible to disconnect. Need either address-based disconnect
       (`disconnect --addr 10.99.1.10:43006`) or a short connection ID shown in
       `wallhack peers` output that can be passed to `disconnect`.
-- [ ] **Stale peer not cleaned up after remote restart** — when a peer process is
-      killed and restarts (e.g. gateway-perimeter exit→relay), the old connection
-      lingers in `wallhack peers` as a second entry for the same host alongside the
-      new connection. Two peers for the same physical node is confusing and indicates
-      the old session wasn't properly torn down.
+- [x] ~~**Stale peer not cleaned up after remote restart**~~ — fixed by
+      connection ID–guarded `unregister_if_current` in PR #82.
 - [ ] **`status=connected latency=— tun=false listen=false connect=false` is an
       impossible state** — `wallhack peers` shows a connected peer with all capability
       flags false and no latency. If `status=connected` then either `listen` or
@@ -105,16 +118,12 @@
       (`peer_name_to_iface`) so TUN gets a random name instead of `wh{hash}`.
 - [ ] **Relay peer role reported as `exit`** — `wallhack peers` shows relay peers
       as `role=exit`. Should report `role=relay`.
-- [ ] **Stale TUN interfaces not cleaned up on disconnect** — when a peer disconnects
-      (or restarts with a different role), the old TUN interface remains on the entry
-      node and kernel routes pointing to it linger. Auto-cleanup on disconnect needed.
-- [ ] TUN EBUSY on rapid reconnect — `create_tun_with_retry` (entry) retries
-      3× at 500ms but the previous `TunActor` hasn't been fully dropped before
-      the new connection attempts to claim the same TUN name. Rapid
-      connect/disconnect cycles accumulate stale connections, eventually causing
-      resource exhaustion and process kill (OOM or SIGKILL). Needs proper TUN
-      lifecycle tracking — ensure the old actor is fully dropped before allowing
-      a new connection to reuse the name.
+- [x] ~~**Stale TUN interfaces not cleaned up on disconnect**~~ — fixed in PR #82:
+      `delete_tun` runs after manager task abort+join; kernel auto-removes routes.
+- [x] ~~TUN EBUSY on rapid reconnect~~ — fixed in PR #82:
+      `SessionManager.get_or_create` preemptively deletes stale TUN;
+      `run_connection_loop` aborts+joins manager task to release fd Arcs;
+      `TunDropGuard` for panic safety.
 - [x] ~~No color in `[+]` notification messages~~ — done, uses `nu-ansi-term`
       behind `repl` feature gate.
 - [ ] Log prefix inconsistency in REPL — mix of `warn:` prefix (from

crates/core/src/control/peers.rs

@@ -2,7 +2,10 @@
 
 use std::{
     collections::HashMap,
-    sync::Arc,
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
     time::{Duration, Instant},
 };
 
@@ -70,6 +73,8 @@ pub struct PeerInfo {
     pub latency_measured_at: Option<Instant>,
     /// TUN interface name for this peer (entry-side only).
     pub tun_name: Option<String>,
+    /// Unique identifier for this connection instance.
+    pub connection_id: u64,
 }
 
 /// Shared peer registry.
@@ -85,6 +90,8 @@ pub struct Registry {
     ping_channels: ArcSwap<HashMap<String, mpsc::Sender<PingRequest>>>,
     /// Broadcast channel for peer lifecycle events.
     events_tx: broadcast::Sender<PeerEvent>,
+    /// Monotonic counter for assigning unique connection IDs.
+    next_connection_id: AtomicU64,
 }
 
 impl Default for Registry {
@@ -94,6 +101,7 @@ impl Default for Registry {
             peers: ArcSwap::from_pointee(HashMap::new()),
             ping_channels: ArcSwap::from_pointee(HashMap::new()),
             events_tx,
+            next_connection_id: AtomicU64::new(0),
         }
     }
 }
@@ -122,7 +130,12 @@ impl Registry {
     }
 
     /// Register a new peer.
-    pub fn register(&self, id: String, addr: String, role: NodeRole, side: ConnectionSide) {
+    ///
+    /// Returns a connection ID that must be passed to
+    /// [`unregister_if_current`] to prevent a stale task from evicting
+    /// a newer connection that re-registered under the same name.
+    pub fn register(&self, id: String, addr: String, role: NodeRole, side: ConnectionSide) -> u64 {
+        let connection_id = self.next_connection_id.fetch_add(1, Ordering::Relaxed) + 1;
         // TOCTOU: another thread could register the same id between this
         // check and the rcu insert. Harmless — a duplicate Connected event
         // is better than a missed one, and callers don't race on the same id.
@@ -138,6 +151,7 @@ impl Registry {
             latency_ms: None,
             latency_measured_at: None,
             tun_name: None,
+            connection_id,
         };
         let event_addr = info.addr.clone();
         let event_role = info.role;
@@ -154,6 +168,7 @@ impl Registry {
                 role: event_role,
             });
         }
+        connection_id
     }
 
     /// Set the TUN interface name for a peer.
@@ -179,7 +194,7 @@ impl Registry {
         });
     }
 
-    /// Unregister a peer.
+    /// Unregister a peer unconditionally.
     pub fn unregister(&self, id: &str) -> Option<PeerInfo> {
         self.ping_channels.rcu(|old| {
             let mut new = (**old).clone();
@@ -200,6 +215,25 @@ impl Registry {
         removed
     }
 
+    /// Unregister a peer only if its connection ID matches.
+    ///
+    /// Prevents a stale task from evicting a newer connection that
+    /// re-registered under the same name between the old task's exit
+    /// and its cleanup.
+    pub fn unregister_if_current(&self, id: &str, connection_id: u64) -> Option<PeerInfo> {
+        let current = self.peers.load().get(id).map(|p| p.connection_id);
+        if current != Some(connection_id) {
+            tracing::debug!(
+                peer = id,
+                current = ?current,
+                requested = connection_id,
+                "skipping stale unregister"
+            );
+            return None;
+        }
+        self.unregister(id)
+    }
+
     /// Register a ping channel for a peer's connection handler.
     ///
     /// Returns the receiver that the connection handler should listen on.
@@ -475,4 +509,31 @@ mod tests {
 
         assert!(rx.try_recv().is_err());
     }
+
+    #[test]
+    fn test_connection_id_prevents_stale_unregister() {
+        let registry = Registry::new();
+        let gen1 = registry.register(
+            "peer1".into(),
+            "1.2.3.4:5678".into(),
+            NodeRole::Exit,
+            ConnectionSide::Accept,
+        );
+        // Re-register same peer (new connection arrived before old task cleaned up).
+        let gen2 = registry.register(
+            "peer1".into(),
+            "1.2.3.4:9999".into(),
+            NodeRole::Exit,
+            ConnectionSide::Accept,
+        );
+        assert_ne!(gen1, gen2);
+
+        // Old task tries to unregister with stale connection_id — should be a no-op.
+        assert!(registry.unregister_if_current("peer1", gen1).is_none());
+        assert_eq!(registry.count(), 1);
+
+        // New task unregisters with current connection_id — should succeed.
+        assert!(registry.unregister_if_current("peer1", gen2).is_some());
+        assert_eq!(registry.count(), 0);
+    }
 }

crates/core/src/entry/manager.rs

@@ -5,6 +5,7 @@ use smoltcp::phy::Device;
 use tokio::{
     io::unix::AsyncFd,
     sync::{Notify, mpsc},
+    task::JoinSet,
     time::Instant,
 };
 use wallhack_entry_stack::async_stack::{
@@ -61,7 +62,7 @@ pub struct ConnectionManager<D: Device + Send + 'static> {
     egress_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
     /// Intercepted ICMP Echo Requests from the poll loop for tunnel forwarding.
     icmp_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
-    /// Unified data stream: send UDP instructions to the exit node.
+    /// Unified data stream: send instructions to the exit node.
     instructions_tx: mpsc::Sender<EntryNodeInstruction>,
     /// Unified data stream: receive responses from the exit node.
     responses_rx: mpsc::Receiver<ExitNodeResponse>,
@@ -116,7 +117,11 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
         let udp_timeout = Duration::from_secs(30);
         let mut udp_buf = vec![0u8; 65535];
 
-        loop {
+        // Legacy probe tasks (fallback bidi-stream probes) in a JoinSet so
+        // they are aborted on exit, releasing Arc<AsyncFd<Device>> clones.
+        let mut probe_tasks: JoinSet<()> = JoinSet::new();
+
+        let result = loop {
             tokio::select! {
                 stream = listener.accept() => {
                     let stream = stream?;
@@ -187,7 +192,7 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                     };
                     if self.instructions_tx.send(instr).await.is_err() {
                         tracing::debug!("UDP: instructions channel closed, stopping");
-                        return Ok(());
+                        break Ok(());
                     }
                     self.metrics.inc_packets_out(1);
                     self.metrics.inc_bytes_out(size as u64);
@@ -197,7 +202,7 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                         self.handle_exit_response(&mut udp, response);
                     } else {
                         tracing::debug!("Responses channel closed, connection dead");
-                        return Ok(());
+                        break Ok(());
                     }
                 }
                 Some(held) = self.syn_rx.recv() => {
@@ -207,7 +212,7 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                     let tun = Arc::clone(&self.tun_writer);
                     let wake = Arc::clone(&self.wake_notify);
                     let dst_addr = held.dst_addr;
-                    tokio::spawn(async move {
+                    probe_tasks.spawn(async move {
                         let Some(target_addr) = parse_syn_target(&held.packet) else {
                             tracing::debug!(%dst_addr, "SYN probe: failed to parse target");
                             state.mark_unreachable(dst_addr);
@@ -240,6 +245,12 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                         wake.notify_one();
                     });
                 }
+                // Reap completed probe tasks so the JoinSet doesn't grow unbounded.
+                Some(result) = probe_tasks.join_next(), if !probe_tasks.is_empty() => {
+                    if let Err(e) = result {
+                        tracing::warn!("SYN probe task panicked: {e}");
+                    }
+                }
                 Some(icmp_packet) = self.egress_rx.recv() => {
                     // ICMP packets from the poll loop (cache-hit retransmits).
                     write_raw_to_tun(&self.tun_writer, &icmp_packet);
@@ -251,7 +262,7 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                         tracing::trace!(len = icmp_pkt.len(), "forwarding ICMP echo request to exit");
                         if self.instructions_tx.send(instr).await.is_err() {
                             tracing::debug!("ICMP: instructions channel closed, stopping");
-                            return Ok(());
+                            break Ok(());
                         }
                         self.metrics.inc_packets_out(1);
                         self.metrics.inc_bytes_out(icmp_pkt.len() as u64);
@@ -271,7 +282,15 @@ impl<D: Device + Send + 'static> ConnectionManager<D> {
                     });
                 }
             }
-        }
+        };
+
+        // Abort in-flight probe tasks and wait for cleanup so their
+        // Arc<AsyncFd<Device>> clones are dropped before the caller
+        // deletes the TUN interface.
+        probe_tasks.abort_all();
+        while probe_tasks.join_next().await.is_some() {}
+
+        result
     }
 
     #[allow(clippy::too_many_lines)]