fix(notifier): lazy-GC dead sessions on every fanout / per-peer push

rmcp's ServerHandler doesn't expose a session-close callback and the
streamable-HTTP session manager owns the close path internally, so we
have no obvious place to call unregister_session. Without GC the
sessions map grows unbounded across the gateway's lifetime — every
reconnect leaves stale entries that fanout iterates and the resolver
attempts to re-route.

What rmcp *does* give us is `Peer<R>::is_transport_closed()`, which
flips true once the underlying transport has terminated. Reap lazily:
each fanout / per-peer push snapshots the live session list, scans for
closed peers, and removes them from both `sessions` and the
`feature_set_resolver`'s `SessionRootsRegistry` in one pass. After the
sweep the regular routing loop runs against the cleaned snapshot.

Three call sites covered: get_peers_for_space (broadcasts),
get_peers_for_space_with_streams (the variant used by the per-type
notify_*_list_changed), and notify_peer_lists_changed (per-client
push for resolution flips and grant edits). Logged at info level
when dead > 0 so a future spike is visible.

Adds `FeatureSetResolverService::session_roots()` accessor so the
notifier can keep the two registries in sync — they were drifting
silently otherwise.

cargo check + clippy (-D warnings) clean.

Signed-off-by: Mohammod Al Amin Ashik <maa.ashik00@gmail.com>

Author: its-mash

crates/mcpmux-gateway/src/consumers/mcp_notifier.rs

@@ -322,13 +322,61 @@ impl MCPNotifier {
         tracker.insert((space_id, NotificationType::All), timestamp);
     }
 
+    /// Lazy GC for dead sessions.
+    ///
+    /// rmcp's `ServerHandler` doesn't expose a session-close callback, and
+    /// the streamable-HTTP session manager owns the close path internally.
+    /// What we *do* have on every `Peer<R>` is `is_transport_closed()` —
+    /// it flips true once the underlying transport has terminated. So we
+    /// reap lazily: every fanout / probe pass scans for closed peers and
+    /// removes them from both `sessions` and `session_roots`.
+    ///
+    /// Returns the ids that were reaped (for logging / metrics). Callers
+    /// pass the live (snapshot) list of `(session_id, peer)` they were
+    /// about to iterate; this mutates `self.sessions` and the
+    /// `feature_set_resolver`'s session registry.
+    fn reap_dead_sessions(&self, snapshot: &[(String, Arc<Peer<RoleServer>>)]) -> Vec<String> {
+        let dead: Vec<String> = snapshot
+            .iter()
+            .filter_map(|(sid, peer)| {
+                if peer.is_transport_closed() {
+                    Some(sid.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
+        if dead.is_empty() {
+            return dead;
+        }
+        {
+            let mut sessions = self.sessions.write();
+            for sid in &dead {
+                sessions.remove(sid);
+            }
+        }
+        // Also clean the session_roots registry the resolver consults so
+        // it doesn't keep returning stale roots / capability flags for
+        // sessions that no longer exist.
+        for sid in &dead {
+            self.feature_set_resolver.session_roots().remove(sid);
+        }
+        info!(
+            reaped = dead.len(),
+            "[MCPNotifier] 🧹 reaped dead sessions (transport closed)"
+        );
+        dead
+    }
+
     /// Get every peer whose **session** currently routes into `space_id`.
     ///
     /// Iterates the session registry and re-runs the FeatureSet resolver
     /// per session — same logic the request handlers use, so a session
     /// redirected by `WorkspaceBinding` to a non-default space is matched
     /// correctly. Sessions whose stream isn't active yet are skipped (the
-    /// notification would be queued but not delivered).
+    /// notification would be queued but not delivered). Dead sessions
+    /// (transport closed) are GC'd before the resolve pass so we don't
+    /// waste a `feature_set_resolver.resolve()` call on them.
     async fn get_peers_for_space(&self, space_id: Uuid) -> Vec<Arc<Peer<RoleServer>>> {
         let session_list: Vec<(String, String, Arc<Peer<RoleServer>>)> = {
             let sessions = self.sessions.read();
@@ -339,9 +387,22 @@ impl MCPNotifier {
                 .collect()
         };
 
+        // GC dead sessions before resolving — also drops them from the
+        // snapshot so the resolve loop below skips them.
+        let dead = self.reap_dead_sessions(
+            &session_list
+                .iter()
+                .map(|(sid, _, peer)| (sid.clone(), peer.clone()))
+                .collect::<Vec<_>>(),
+        );
+        let dead_set: std::collections::HashSet<&str> = dead.iter().map(String::as_str).collect();
+
         let mut matching_peers = Vec::new();
         let _space_resolver = &self.space_resolver; // kept-but-unused; resolver below is authoritative
         for (session_id, client_id, peer) in session_list {
+            if dead_set.contains(session_id.as_str()) {
+                continue;
+            }
             match self
                 .feature_set_resolver
                 .resolve(Some(&session_id), Some(&client_id))
@@ -811,10 +872,21 @@ impl MCPNotifier {
                 .collect()
         };
 
+        let dead = self.reap_dead_sessions(
+            &session_list
+                .iter()
+                .map(|(sid, _, peer)| (sid.clone(), peer.clone()))
+                .collect::<Vec<_>>(),
+        );
+        let dead_set: std::collections::HashSet<&str> = dead.iter().map(String::as_str).collect();
+
         let mut matching_peers = Vec::new();
         let mut matching_client_ids = Vec::new();
 
         for (session_id, client_id, peer) in session_list {
+            if dead_set.contains(session_id.as_str()) {
+                continue;
+            }
             match self
                 .feature_set_resolver
                 .resolve(Some(&session_id), Some(&client_id))
@@ -995,14 +1067,21 @@ impl MCPNotifier {
         // editors, parallel CLI invocations). Push the notification on
         // every active session for that client_id; client-side dedup is
         // their problem, but missing a session would be ours.
-        let peers: Vec<Arc<Peer<RoleServer>>> = {
+        let snapshot: Vec<(String, Arc<Peer<RoleServer>>)> = {
             let sessions = self.sessions.read();
             sessions
                 .iter()
                 .filter(|(_, e)| e.client_id == client_id && e.has_active_stream)
-                .map(|(_, e)| e.peer.clone())
+                .map(|(sid, e)| (sid.clone(), e.peer.clone()))
                 .collect()
         };
+        let dead = self.reap_dead_sessions(&snapshot);
+        let dead_set: std::collections::HashSet<&str> = dead.iter().map(String::as_str).collect();
+        let peers: Vec<Arc<Peer<RoleServer>>> = snapshot
+            .into_iter()
+            .filter(|(sid, _)| !dead_set.contains(sid.as_str()))
+            .map(|(_, peer)| peer)
+            .collect();
 
         if peers.is_empty() {
             debug!(

crates/mcpmux-gateway/src/services/feature_set_resolver.rs

@@ -118,6 +118,13 @@ impl FeatureSetResolverService {
         }
     }
 
+    /// Borrow the session-roots registry. The notifier uses this to GC
+    /// dead sessions out of the registry when reaping the corresponding
+    /// peer entries — keeping both stores in sync.
+    pub fn session_roots(&self) -> &Arc<SessionRootsRegistry> {
+        &self.session_roots
+    }
+
     /// Resolve the effective (Space, FS list, source) tuple for a session.
     ///
     /// `session_id`: the client's `mcp-session-id` header (or `None` when