Preserve live peer sessions across equivalent route revision bumps

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift

@@ -25,6 +25,7 @@ public actor CmxConnectivityEngine {
     private var endpointGeneration: UInt64?
     private var localIdentity: CmxIrohPeerIdentity?
     private var routeRevision: UInt64?
+    private var routeContent: CmxConnectivityRouteContent?
     private var endpointEventTask: Task<Void, Never>?
     private var routeSyncOperation: RouteSyncOperation?
     private var peers: [CmxConnectivityPeerID: CmxConnectivityPeerSession] = [:]
@@ -240,10 +241,22 @@ public actor CmxConnectivityEngine {
     }
 
     /// Records the last route revision installed atomically by the composition root.
-    public func didInstallRouteRevision(_ revision: UInt64) async {
-        guard routeRevision != revision else { return }
-        await invalidateAllPeers(failure: .superseded)
+    ///
+    /// Peers whose material route content is unchanged keep their live
+    /// sessions; every other peer is invalidated before the new revision
+    /// becomes visible.
+    public func didInstallRouteRevision(
+        _ revision: UInt64,
+        routes: CmxIrohDiscoveryResponse
+    ) async {
+        let content = CmxConnectivityRouteContent(snapshot: routes)
+        guard routeRevision != revision else {
+            routeContent = content
+            return
+        }
+        await invalidatePeersSuperseded(by: content)
         routeRevision = revision
+        routeContent = content
         publishSnapshot()
     }
 
@@ -678,10 +691,36 @@ public actor CmxConnectivityEngine {
                 throw CmxConnectivityEngineError.superseded
             }
         }
+        let content = response.snapshot.map(CmxConnectivityRouteContent.init)
         if routeRevision != response.revision {
-            await invalidateAllPeers(failure: .superseded)
+            await invalidatePeersSuperseded(by: content)
             routeRevision = response.revision
+            routeContent = content
             publishSnapshot()
+        } else if let content {
+            routeContent = content
+        }
+    }
+
+    /// Invalidates peers whose authoritative route material changed.
+    ///
+    /// A missing baseline or replacement fails closed and tears down every
+    /// peer, preserving the pre-content-tracking behavior.
+    private func invalidatePeersSuperseded(
+        by content: CmxConnectivityRouteContent?
+    ) async {
+        guard let previous = routeContent,
+              let content,
+              previous.account == content.account else {
+            await invalidateAllPeers(failure: .superseded)
+            return
+        }
+        for (peerID, peer) in peers {
+            guard let previousRoute = previous.peerRoute(for: peerID),
+                  previousRoute == content.peerRoute(for: peerID) else {
+                await peer.invalidate(failure: .superseded)
+                continue
+            }
         }
     }
 

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift

@@ -200,6 +200,16 @@ actor CmxConnectivityPeerSession {
                 continue redial
             }
 
+            // The dead-on-arrival probe suspends this actor. A concurrent
+            // caller that dialed in that window may have installed first;
+            // installing over it would leak its session and double-record
+            // an established lifecycle for the same peer.
+            if let installed = activeConnection {
+                if installed.id != pending.id {
+                    await connected.close()
+                }
+                return installed.session
+            }
             install(
                 connected,
                 id: pending.id,

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift

@@ -0,0 +1,65 @@
+/// Authoritative route material whose change requires live session teardown.
+///
+/// Volatile freshness fields are excluded on purpose: `last_seen_at`, path
+/// hints, direct ports, and display names move on every registration
+/// heartbeat and shape only the next dial, never the trust of an already
+/// admitted connection. Comparing this content lets a route revision bump
+/// keep healthy sessions whose routes did not materially change.
+struct CmxConnectivityRouteContent: Equatable, Sendable {
+    /// Trust material shared by every route in one account snapshot.
+    struct AccountMaterial: Equatable, Sendable {
+        let relayFleet: [String]
+        let lanRendezvous: CmxIrohLANRendezvous
+        let grantVerificationKeys: CmxIrohGrantVerificationKeySet
+    }
+
+    /// Admission-relevant material of one broker binding.
+    struct BindingMaterial: Equatable, Sendable {
+        let bindingID: String
+        let appInstanceID: String
+        let tag: String
+        let platform: CmxIrohPlatform
+        let identityGeneration: Int
+        let pairingEnabled: Bool
+        let capabilities: [String]
+
+        init(binding: CmxIrohBrokerBinding) {
+            bindingID = binding.bindingID
+            appInstanceID = binding.appInstanceID
+            tag = binding.tag
+            platform = binding.platform
+            identityGeneration = binding.identityGeneration
+            pairingEnabled = binding.pairingEnabled
+            capabilities = binding.capabilities
+        }
+    }
+
+    let account: AccountMaterial
+    private let peerRoutes: [CmxConnectivityPeerID: [BindingMaterial]]
+
+    init(snapshot: CmxIrohDiscoveryResponse) {
+        account = AccountMaterial(
+            relayFleet: snapshot.relayFleet,
+            lanRendezvous: snapshot.lanRendezvous,
+            grantVerificationKeys: snapshot.grantVerificationKeys
+        )
+        var routes: [CmxConnectivityPeerID: [BindingMaterial]] = [:]
+        for binding in snapshot.bindings {
+            let peerID = CmxConnectivityPeerID(
+                identity: binding.endpointID,
+                deviceID: binding.deviceID
+            )
+            routes[peerID, default: []].append(BindingMaterial(binding: binding))
+        }
+        peerRoutes = routes.mapValues { bindings in
+            bindings.sorted { $0.bindingID < $1.bindingID }
+        }
+    }
+
+    /// Returns the material route for one peer, or nil when unrouted.
+    func peerRoute(
+        for peerID: CmxConnectivityPeerID
+    ) -> [BindingMaterial]? {
+        peerRoutes[peerID]
+    }
+}

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime+PolicyRefresh.swift

@@ -88,7 +88,10 @@ extension CmxIrohClientRuntime {
                 try requireCurrent(revision)
                 guard published else { return .failed(.superseded) }
                 if let routeRevision = discovery.revision {
-                    await connectivityEngine.didInstallRouteRevision(routeRevision)
+                    await connectivityEngine.didInstallRouteRevision(
+                        routeRevision,
+                        routes: discovery
+                    )
                 }
                 liveDiscoveryGeneration &+= 1
                 return .refreshed

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime.swift

@@ -376,7 +376,10 @@ public actor CmxIrohClientRuntime {
             guard published else {
                 return .failed(.superseded)
             }
-            await connectivityEngine.didInstallRouteRevision(discoveredRevision)
+            await connectivityEngine.didInstallRouteRevision(
+                discoveredRevision,
+                routes: discovery
+            )
             liveDiscoveryGeneration &+= 1
             return .refreshed
         } catch {
@@ -520,7 +523,8 @@ public actor CmxIrohClientRuntime {
                 if published {
                     if let routeRevision = discovery.revision {
                         await connectivityEngine.didInstallRouteRevision(
-                            routeRevision
+                            routeRevision,
+                            routes: discovery
                         )
                     }
                     liveDiscoveryGeneration &+= 1

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PolicyRefresh.swift

@@ -459,7 +459,10 @@ extension CmxIrohHostRuntime {
             await handleRoute(policy.binding, policy.routePathHints)
             try requireCurrent(revision)
             if let routeRevision = discovery.revision {
-                await connectivityEngine.didInstallRouteRevision(routeRevision)
+                await connectivityEngine.didInstallRouteRevision(
+                    routeRevision,
+                    routes: discovery
+                )
             }
             scheduleLANPublication(
                 binding: policy.binding,

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PublicAPI.swift

@@ -149,7 +149,10 @@ extension CmxIrohHostRuntime {
             try requireCurrent(revision)
             await handleRoute(metadata, discovered.pathHints)
             try requireCurrent(revision)
-            await connectivityEngine.didInstallRouteRevision(discoveredRevision)
+            await connectivityEngine.didInstallRouteRevision(
+                discoveredRevision,
+                routes: discovery
+            )
             scheduleLANPublication(
                 binding: metadata,
                 rendezvous: discovery.lanRendezvous,

Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime.swift

@@ -321,7 +321,10 @@ public actor CmxIrohHostRuntime {
                 await handleBinding(registration, discovery, publishedPolicy.attestation)
                 try requireCurrent(revision)
                 if let routeRevision = discovery.revision {
-                    await connectivityEngine.didInstallRouteRevision(routeRevision)
+                    await connectivityEngine.didInstallRouteRevision(
+                        routeRevision,
+                        routes: discovery
+                    )
                 }
                 scheduleRegistrationRenewal(
                     binding: registration.binding,