fix adapter cancellation ownership races

crates/iroh-http-core/src/endpoint/bind.rs

@@ -261,7 +261,7 @@ impl IrohEndpoint {
                 closed_rx,
                 event_tx,
                 event_rx: std::sync::Mutex::new(Some(event_rx)),
-                path_subs: dashmap::DashMap::new(),
+                path_subs: std::sync::Mutex::new(std::collections::HashMap::new()),
                 active_path_watchers: std::sync::atomic::AtomicUsize::new(0),
             },
             ffi: FfiBridge {

crates/iroh-http-core/src/endpoint/observe.rs

@@ -8,6 +8,7 @@ use std::sync::atomic::Ordering;
 use iroh::endpoint::TransportAddrUsage;
 
 use super::{
+    session_runtime::PathSubscriptions,
     stats::{EndpointStats, NodeAddrInfo, PathInfo, PeerStats},
     IrohEndpoint,
 };
@@ -22,7 +23,13 @@ impl IrohEndpoint {
         let pool_size = self.inner.http.pool.entry_count_approx() as usize;
         let active_connections = self.inner.http.active_connections.load(Ordering::Relaxed);
         let active_requests = self.inner.http.active_requests.load(Ordering::Relaxed);
-        let active_path_subscriptions = self.inner.session.path_subs.len();
+        let active_path_subscriptions = self
+            .inner
+            .session
+            .path_subs
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .len();
         let active_path_watchers = self
             .inner
             .session
@@ -232,23 +239,40 @@ impl IrohEndpoint {
     /// The watcher polls `peer_stats()` every 200 ms and emits on the returned
     /// channel whenever the active path changes.
     ///
-    /// Re-subscribing to a peer that already has a live watcher reuses that
-    /// watcher: only the sender is swapped, so no additional task is spawned and
-    /// `active_path_watchers` still counts exactly one live watcher per peer.
+    /// Additional subscriptions to a peer reuse its watcher and receive the
+    /// same changes. `subscription_id` scopes cancellation to its own receiver.
     pub fn subscribe_path_changes(
         &self,
         node_id_str: &str,
+        subscription_id: u32,
     ) -> tokio::sync::mpsc::UnboundedReceiver<PathInfo> {
         let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
-        // Replace any existing sender; the running watcher (if any) picks up the
-        // new sender on its next poll. `insert` returns the previous sender when
-        // a watcher is already live for this peer.
-        let had_existing_watcher = self
-            .inner
-            .session
-            .path_subs
-            .insert(node_id_str.to_string(), tx)
-            .is_some();
+        let (peer_subscriptions, had_existing_watcher) = {
+            let mut subscriptions = self
+                .inner
+                .session
+                .path_subs
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            let (peer_subscriptions, had_existing) = match subscriptions
+                .entry(node_id_str.to_string())
+            {
+                std::collections::hash_map::Entry::Occupied(entry) => (entry.get().clone(), true),
+                std::collections::hash_map::Entry::Vacant(entry) => {
+                    let peer_subscriptions = std::sync::Arc::new(PathSubscriptions {
+                        senders: std::sync::Mutex::new(std::collections::HashMap::new()),
+                    });
+                    entry.insert(peer_subscriptions.clone());
+                    (peer_subscriptions, false)
+                }
+            };
+            peer_subscriptions
+                .senders
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .insert(subscription_id, tx);
+            (peer_subscriptions, had_existing)
+        };
 
         // A watcher already exists for this peer — reuse it. Spawning another
         // would leak a task and over-count `active_path_watchers`.
@@ -270,18 +294,47 @@ impl IrohEndpoint {
             loop {
                 // Exit immediately if the endpoint has been closed.
                 if *closed_rx.borrow() {
-                    ep.inner.session.path_subs.remove(&nid);
+                    let mut subscriptions = ep
+                        .inner
+                        .session
+                        .path_subs
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner);
+                    if subscriptions
+                        .get(&nid)
+                        .is_some_and(|current| std::sync::Arc::ptr_eq(current, &peer_subscriptions))
+                    {
+                        subscriptions.remove(&nid);
+                    }
                     break;
                 }
-                let is_closed = ep
-                    .inner
-                    .session
-                    .path_subs
-                    .get(&nid)
-                    .map(|s| s.is_closed())
-                    .unwrap_or(true);
+                let is_closed = {
+                    let mut subscriptions = ep
+                        .inner
+                        .session
+                        .path_subs
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner);
+                    let Some(current) = subscriptions.get(&nid) else {
+                        break;
+                    };
+                    if !std::sync::Arc::ptr_eq(current, &peer_subscriptions) {
+                        break;
+                    }
+                    let is_empty = {
+                        let mut senders = peer_subscriptions
+                            .senders
+                            .lock()
+                            .unwrap_or_else(std::sync::PoisonError::into_inner);
+                        senders.retain(|_, sender| !sender.is_closed());
+                        senders.is_empty()
+                    };
+                    if is_empty {
+                        subscriptions.remove(&nid);
+                    }
+                    is_empty
+                };
                 if is_closed {
-                    ep.inner.session.path_subs.remove(&nid);
                     break;
                 }
 
@@ -290,8 +343,22 @@ impl IrohEndpoint {
                         let key = format!("{}:{}", active.relay, active.addr);
                         if Some(&key) != last_key.as_ref() {
                             last_key = Some(key);
-                            if let Some(sender) = ep.inner.session.path_subs.get(&nid) {
-                                let _ = sender.send(active.clone());
+                            let subscriptions = ep
+                                .inner
+                                .session
+                                .path_subs
+                                .lock()
+                                .unwrap_or_else(std::sync::PoisonError::into_inner);
+                            if subscriptions.get(&nid).is_some_and(|current| {
+                                std::sync::Arc::ptr_eq(current, &peer_subscriptions)
+                            }) {
+                                let senders = peer_subscriptions
+                                    .senders
+                                    .lock()
+                                    .unwrap_or_else(std::sync::PoisonError::into_inner);
+                                for sender in senders.values() {
+                                    let _ = sender.send(active.clone());
+                                }
                             }
                             let _ = event_tx.try_send(
                                 crate::http::events::TransportEvent::path_change(
@@ -309,7 +376,16 @@ impl IrohEndpoint {
                     _ = tokio::time::sleep(std::time::Duration::from_millis(200)) => {}
                     result = closed_rx.wait_for(|v| *v) => {
                         let _ = result;
-                        ep.inner.session.path_subs.remove(&nid);
+                        let mut subscriptions = ep.inner
+                            .session
+                            .path_subs
+                            .lock()
+                            .unwrap_or_else(std::sync::PoisonError::into_inner);
+                        if subscriptions.get(&nid).is_some_and(|current| {
+                            std::sync::Arc::ptr_eq(current, &peer_subscriptions)
+                        }) {
+                            subscriptions.remove(&nid);
+                        }
                         break;
                     }
                 }
@@ -324,8 +400,20 @@ impl IrohEndpoint {
     }
 
     /// Stop watching path changes for a specific peer.
-    pub fn unsubscribe_path_changes(&self, node_id_str: &str) {
-        self.inner.session.path_subs.remove(node_id_str);
+    pub fn unsubscribe_path_changes(&self, node_id_str: &str, subscription_id: u32) {
+        let subscriptions = self
+            .inner
+            .session
+            .path_subs
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if let Some(peer_subscriptions) = subscriptions.get(node_id_str) {
+            peer_subscriptions
+                .senders
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .remove(&subscription_id);
+        }
     }
 }
 

crates/iroh-http-core/src/endpoint/session_runtime.rs

@@ -5,10 +5,10 @@
 //! SessionRuntime intentionally stays here alongside IrohEndpoint (tight
 //! lifecycle coupling; no further move is planned).
 
+use std::collections::HashMap;
 use std::sync::atomic::{AtomicU64, AtomicUsize};
 use std::sync::Mutex;
 
-use dashmap::DashMap;
 use tokio::sync::{mpsc, watch};
 
 use crate::http::events::TransportEvent;
@@ -17,6 +17,10 @@ use crate::http::server::ServeHandle;
 
 use super::stats::PathInfo;
 
+pub(in crate::endpoint) struct PathSubscriptions {
+    pub(in crate::endpoint) senders: Mutex<HashMap<u32, mpsc::UnboundedSender<PathInfo>>>,
+}
+
 /// Server-side runtime: the `serve()` task, lifecycle signals, and
 /// observability fan-out (transport events, per-peer path subscriptions).
 pub(in crate::endpoint) struct SessionRuntime {
@@ -43,9 +47,10 @@ pub(in crate::endpoint) struct SessionRuntime {
     /// Receiver for transport-level events. Wrapped in Mutex+Option so
     /// `subscribe_events()` can take it exactly once for the platform drain task.
     pub(in crate::endpoint) event_rx: Mutex<Option<mpsc::Receiver<TransportEvent>>>,
-    /// Per-peer path-change subscriptions. Key: `node_id_str`. Populated
-    /// lazily when `subscribe_path_changes` is called.
-    pub(in crate::endpoint) path_subs: DashMap<String, mpsc::UnboundedSender<PathInfo>>,
+    /// Per-peer path-change subscriptions. The inner key is an adapter-owned
+    /// subscription ID, allowing one watcher to fan out without one iterator's
+    /// cancellation terminating another.
+    pub(in crate::endpoint) path_subs: Mutex<HashMap<String, std::sync::Arc<PathSubscriptions>>>,
     /// Number of live path-change watcher tasks.
     pub(in crate::endpoint) active_path_watchers: AtomicUsize,
 }

crates/iroh-http-core/tests/observe.rs

@@ -24,18 +24,18 @@ async fn bind_disabled() -> IrohEndpoint {
 /// did not stop the previous watcher (it re-read the map key and saw the new,
 /// non-closed sender), so old watchers never exited.
 ///
-/// Fix: reuse the existing watcher for a peer that is already subscribed —
-/// replace only the sender, and neither spawn a new task nor bump the gauge.
+/// Fix: reuse the existing watcher for a peer that is already subscribed and
+/// fan out to token-owned senders without spawning another watcher.
 #[tokio::test]
 async fn repeated_subscribe_same_peer_counts_one_watcher() {
     let ep = bind_disabled().await;
     let peer = "test-peer-node-id";
 
     // Subscribe three times for the same peer. Hold every receiver alive so
     // none of the senders report `is_closed()`.
-    let _rx1 = ep.subscribe_path_changes(peer);
-    let _rx2 = ep.subscribe_path_changes(peer);
-    let _rx3 = ep.subscribe_path_changes(peer);
+    let _rx1 = ep.subscribe_path_changes(peer, 1);
+    let _rx2 = ep.subscribe_path_changes(peer, 2);
+    let _rx3 = ep.subscribe_path_changes(peer, 3);
 
     let stats = ep.endpoint_stats();
     assert_eq!(
@@ -58,14 +58,24 @@ async fn unsubscribe_decrements_watcher_gauge_to_zero() {
     let ep = bind_disabled().await;
     let peer = "test-peer-node-id";
 
-    let rx1 = ep.subscribe_path_changes(peer);
-    let rx2 = ep.subscribe_path_changes(peer);
+    let rx1 = ep.subscribe_path_changes(peer, 1);
+    let mut rx2 = ep.subscribe_path_changes(peer, 2);
     assert_eq!(ep.endpoint_stats().active_path_watchers, 1);
 
-    // Drop receivers and unsubscribe so the watcher observes closure and exits.
+    // Cancelling one token must leave the peer's other subscription alive.
     drop(rx1);
+    ep.unsubscribe_path_changes(peer, 1);
+    assert!(
+        matches!(
+            rx2.try_recv(),
+            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
+        ),
+        "unsubscribing one token must not disconnect another"
+    );
+
+    // Drop the remaining receiver and unsubscribe so the watcher exits.
     drop(rx2);
-    ep.unsubscribe_path_changes(peer);
+    ep.unsubscribe_path_changes(peer, 2);
 
     // Watcher wakes at most every ~200 ms; poll until it exits.
     let mut watchers = usize::MAX;
@@ -83,3 +93,41 @@ async fn unsubscribe_decrements_watcher_gauge_to_zero() {
 
     ep.close().await;
 }
+
+/// A last-token unsubscribe followed immediately by a new token must reuse the
+/// watcher that has not yet observed the empty subscriber set.
+#[tokio::test]
+async fn rapid_resubscribe_does_not_duplicate_peer_watcher() {
+    let ep = bind_disabled().await;
+    let peer = "test-peer-node-id";
+
+    let rx1 = ep.subscribe_path_changes(peer, 1);
+    drop(rx1);
+    ep.unsubscribe_path_changes(peer, 1);
+
+    let mut rx2 = ep.subscribe_path_changes(peer, 2);
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert_eq!(
+        ep.endpoint_stats().active_path_watchers,
+        1,
+        "rapid resubscribe must reuse the existing peer watcher"
+    );
+    assert!(
+        matches!(
+            rx2.try_recv(),
+            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
+        ),
+        "replacement subscription must remain connected"
+    );
+
+    drop(rx2);
+    ep.unsubscribe_path_changes(peer, 2);
+    for _ in 0..50 {
+        if ep.endpoint_stats().active_path_watchers == 0 {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert_eq!(ep.endpoint_stats().active_path_watchers, 0);
+    ep.close().await;
+}

packages/iroh-http-deno/src/adapter.ts

@@ -1569,20 +1569,24 @@ export class DenoAdapter extends IrohAdapter {
   override async nextPathChange(
     _endpointHandle: number,
     nodeId: string,
+    subscriptionId: number,
   ): Promise<PathInfo | null> {
     return call<PathInfo | null>("nextPathChange", {
       endpointHandle: this.#eh,
       nodeId,
+      subscriptionId,
     });
   }
 
   override async unsubscribePathChanges(
     _endpointHandle: number,
     nodeId: string,
+    subscriptionId: number,
   ): Promise<void> {
     await call<null>("unsubscribePathChanges", {
       endpointHandle: this.#eh,
       nodeId,
+      subscriptionId,
     });
   }
 }