diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift index a1287fed5d7d..cfd333b05ef9 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift +++ b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift @@ -244,14 +244,24 @@ public actor CmxConnectivityEngine { /// /// Peers whose material route content is unchanged keep their live /// sessions; every other peer is invalidated before the new revision - /// becomes visible. + /// becomes visible. Account route revisions are monotonic, so an older + /// completion of an overlapping reconciliation cannot roll back a newer + /// installed revision or its content baseline. public func didInstallRouteRevision( _ revision: UInt64, routes: CmxIrohDiscoveryResponse ) async { + if let routeRevision, revision < routeRevision { return } let content = CmxConnectivityRouteContent(snapshot: routes) guard routeRevision != revision else { - routeContent = content + // The recorded revision can lack a content baseline when a sync + // stored it from an unchanged response without a snapshot. A + // missing or differing baseline fails closed like any other + // material change before the content becomes the baseline. + if routeContent != content { + await invalidatePeersSuperseded(by: content) + routeContent = content + } return } await invalidatePeersSuperseded(by: content) diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift index fe90245f74a9..d0242cfbecde 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift +++ b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift @@ -186,10 +186,16 @@ actor CmxConnectivityPeerSession { } if let installed = activeConnection { - if installed.id != pending.id { - await connected.close() + if installed.id == pending.id { + return installed.session + } + if let winner = await settleRedundantDial( + connected, + installedID: installed.id + ) { + return winner } - return installed.session + continue redial } if await connected.isClosed() { await connected.close() @@ -205,10 +211,16 @@ actor CmxConnectivityPeerSession { // 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() + if installed.id == pending.id { + return installed.session + } + if let winner = await settleRedundantDial( + connected, + installedID: installed.id + ) { + return winner } - return installed.session + continue redial } install( connected, @@ -278,6 +290,25 @@ actor CmxConnectivityPeerSession { publishSnapshot() } + /// Closes a redundant dial that lost to an installed winner. + /// + /// Closing suspends this actor, so the winner can be invalidated, + /// replaced, or remotely closed before the close settles. Only a + /// still-installed live winner may be handed out; a nil result means + /// the caller must redial. + private func settleRedundantDial( + _ connected: any CmxConnectivitySession, + installedID: UUID + ) async -> (any CmxConnectivitySession)? { + await connected.close() + guard let current = activeConnection, + current.id == installedID, + !(await current.session.isClosed()) else { + return nil + } + return current.session + } + private func install( _ connected: any CmxConnectivitySession, id: UUID, diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift index 8bdcc4a90530..de7436500e53 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift +++ b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift @@ -7,10 +7,28 @@ /// keep healthy sessions whose routes did not materially change. struct CmxConnectivityRouteContent: Equatable, Sendable { /// Trust material shared by every route in one account snapshot. + /// + /// Relay fleet and verification key order carries no trust meaning, so + /// both are canonicalized here and a reorder-only revision compares + /// equal to the installed material. struct AccountMaterial: Equatable, Sendable { let relayFleet: [String] let lanRendezvous: CmxIrohLANRendezvous let grantVerificationKeys: CmxIrohGrantVerificationKeySet + + init(snapshot: CmxIrohDiscoveryResponse) { + relayFleet = snapshot.relayFleet.sorted() + lanRendezvous = snapshot.lanRendezvous + let keySet = snapshot.grantVerificationKeys + grantVerificationKeys = CmxIrohGrantVerificationKeySet( + version: keySet.version, + currentKeyID: keySet.currentKeyID, + keys: keySet.keys.sorted { + ($0.kid, $0.alg, $0.spkiDerBase64) + < ($1.kid, $1.alg, $1.spkiDerBase64) + } + ) + } } /// Admission-relevant material of one broker binding. @@ -30,7 +48,8 @@ struct CmxConnectivityRouteContent: Equatable, Sendable { platform = binding.platform identityGeneration = binding.identityGeneration pairingEnabled = binding.pairingEnabled - capabilities = binding.capabilities + // The admission policy reads capabilities with set semantics. + capabilities = binding.capabilities.sorted() } } @@ -38,11 +57,7 @@ struct CmxConnectivityRouteContent: Equatable, Sendable { private let peerRoutes: [CmxConnectivityPeerID: [BindingMaterial]] init(snapshot: CmxIrohDiscoveryResponse) { - account = AccountMaterial( - relayFleet: snapshot.relayFleet, - lanRendezvous: snapshot.lanRendezvous, - grantVerificationKeys: snapshot.grantVerificationKeys - ) + account = AccountMaterial(snapshot: snapshot) var routes: [CmxConnectivityPeerID: [BindingMaterial]] = [:] for binding in snapshot.bindings { let peerID = CmxConnectivityPeerID( diff --git a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift index e3a492d8f777..e79d16e9ccdc 100644 --- a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift +++ b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift @@ -286,6 +286,186 @@ struct CmxConnectivityEngineTests { await rig.engine.stop() } + @Test + func reorderedCapabilitiesOnRevisionBumpKeepsTheLivePeerSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z", + capabilities: ["artifact", "terminal"] + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + capabilities: ["terminal", "artifact"] + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.engine.snapshot().routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + await rig.engine.stop() + } + + @Test + func reorderedRelayFleetOnRevisionBumpKeepsTheLivePeerSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z", + relayFleet: [ + "https://relay-a.example/", + "https://relay-b.example/", + ] + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + relayFleet: [ + "https://relay-b.example/", + "https://relay-a.example/", + ] + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.engine.snapshot().routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + await rig.engine.stop() + } + + @Test + func reorderedGrantVerificationKeysOnRevisionBumpKeepsTheLivePeerSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z", + grantVerificationKeyIDs: ["current", "previous"] + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + grantVerificationKeyIDs: ["previous", "current"] + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.engine.snapshot().routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + await rig.engine.stop() + } + + @Test + func snapshotInstallForARevisionRecordedWithoutContentFailsClosed() async throws { + let rig = try await Self.admittedPeerRig( + responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.unchangedResponse(revision: 12), + ], + dialableConnections: 2 + ) + let first = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + try await rig.engine.reconcileRoutes() + #expect(await first.isClosed()) + let second = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + #expect(await second.isClosed() == false) + let snapshot = try #require(Self.peerRouteResponse( + revision: 12, + lastSeenAt: "2026-07-30T00:00:45Z" + ).snapshot) + + await rig.engine.didInstallRouteRevision(12, routes: snapshot) + + #expect(await rig.engine.snapshot().routeRevision == 12) + #expect(await rig.connections[1].observedCloseCallCount() == 1) + #expect(await second.isClosed()) + await rig.engine.stop() + } + + @Test + func sameRevisionReinstallWithUnchangedContentKeepsTheLivePeerSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + let snapshot = try #require(Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z" + ).snapshot) + + await rig.engine.didInstallRouteRevision(10, routes: snapshot) + await rig.engine.didInstallRouteRevision(10, routes: snapshot) + + #expect(await rig.engine.snapshot().routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + await rig.engine.stop() + } + + @Test + func olderRouteRevisionInstallCannotRollBackANewerInstall() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + let newer = try #require(Self.peerRouteResponse( + revision: 11, + lastSeenAt: "2026-07-30T00:00:45Z" + ).snapshot) + let older = try #require(Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:30Z", + identityGeneration: 2 + ).snapshot) + + await rig.engine.didInstallRouteRevision(11, routes: newer) + await rig.engine.didInstallRouteRevision(10, routes: older) + + #expect(await rig.engine.snapshot().routeRevision == 11) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + await rig.engine.stop() + } + @Test func stopFinishesNetworkChangeObservers() async throws { let identity = try CmxIrohPeerIdentity( @@ -317,34 +497,38 @@ struct CmxConnectivityEngineTests { private struct AdmittedPeerRig { let engine: CmxConnectivityEngine - let connection: TestIrohConnection + let connections: [TestIrohConnection] let authority: ScriptedConnectivityAuthority let request: CmxByteTransportRequest + + var connection: TestIrohConnection { connections[0] } } private static func admittedPeerRig( - responses: [CmxConnectivitySyncResponse] + responses: [CmxConnectivitySyncResponse], + dialableConnections: Int = 1 ) async throws -> AdmittedPeerRig { let localIdentity = try CmxIrohPeerIdentity( endpointID: String(repeating: "1", count: 64) ) let peerIdentity = try CmxIrohPeerIdentity(endpointID: peerEndpointID) - let control = CmxIrohBidirectionalStream( - receiveStream: TestIrohReceiveStream( - buffer: CmxIrohAdmissionAckCodec() - .encodeFrame(.acceptedPendingNatTraversal) - + admissionFrame(status: 3) - ), - sendStream: TestIrohSendStream() - ) - let connection = TestIrohConnection( - remoteIdentity: peerIdentity, - bidirectionalStreams: [control], - selectedPath: .direct - ) + let connections = (0 ..< dialableConnections).map { _ in + TestIrohConnection( + remoteIdentity: peerIdentity, + bidirectionalStreams: [CmxIrohBidirectionalStream( + receiveStream: TestIrohReceiveStream( + buffer: CmxIrohAdmissionAckCodec() + .encodeFrame(.acceptedPendingNatTraversal) + + admissionFrame(status: 3) + ), + sendStream: TestIrohSendStream() + )], + selectedPath: .direct + ) + } let endpoint = TestDialingIrohEndpoint( localIdentity: localIdentity, - dialResults: [.connection(connection)] + dialResults: connections.map { .connection($0) } ) let supervisor = CmxIrohEndpointSupervisor( factory: TestIrohEndpointFactory(endpoints: [endpoint]), @@ -373,7 +557,7 @@ struct CmxConnectivityEngineTests { ) return AdmittedPeerRig( engine: engine, - connection: connection, + connections: connections, authority: authority, request: request ) @@ -384,8 +568,13 @@ struct CmxConnectivityEngineTests { lastSeenAt: String, identityGeneration: Int = 1, relayFleet: [String] = ["https://relay.example/"], + capabilities: [String] = ["terminal"], + grantVerificationKeyIDs: [String] = [], includesPeerBinding: Bool = true ) throws -> CmxConnectivitySyncResponse { + let capabilityList = capabilities + .map { "\"\($0)\"" } + .joined(separator: ", ") let binding = """ { "binding_id": "0a0a0a0a-0000-4000-8000-000000000001", @@ -396,7 +585,7 @@ struct CmxConnectivityEngineTests { "endpoint_id": "\(peerEndpointID)", "identity_generation": \(identityGeneration), "pairing_enabled": true, - "capabilities": ["terminal"], + "capabilities": [\(capabilityList)], "path_hints": [], "last_seen_at": "\(lastSeenAt)" } @@ -404,6 +593,13 @@ struct CmxConnectivityEngineTests { let fleet = relayFleet .map { "\"\($0)\"" } .joined(separator: ", ") + let keys = grantVerificationKeyIDs + .map { + """ + {"kid": "\($0)", "alg": "ed25519", "spki_der_base64": "QUJD"} + """ + } + .joined(separator: ", ") return try decodeResponse( """ { @@ -423,7 +619,7 @@ struct CmxConnectivityEngineTests { "grant_verification_keys": { "version": 1, "current_kid": "current", - "keys": [] + "keys": [\(keys)] } } } diff --git a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift index b39854ef2c84..275d1e02ecb2 100644 --- a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift +++ b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift @@ -266,6 +266,111 @@ struct CmxConnectivityPeerSessionTests { await peer.invalidate() } + @Test + func invalidationDuringRedundantDialCloseTriggersAFreshDial() async throws { + let request = try Self.request() + let peerID = try CmxConnectivityPeerID(request: request) + let winner = TestConnectivitySession( + continuityID: 91, + gatesFirstIsClosedCheck: true + ) + let loser = TestConnectivitySession( + continuityID: 92, + gatesFirstClose: true + ) + let replacement = TestConnectivitySession(continuityID: 93) + let builder = OrderedGatedConnectivitySessionBuilder( + sessions: [winner, loser, replacement] + ) + let peer = CmxConnectivityPeerSession( + peerID: peerID, + buildSession: { request in + try await builder.build(request) + } + ) + + // Park the first caller at the dead-on-arrival probe so the second + // caller starts its own dial, then let the winner install before the + // second dial resolves. + let firstCaller = Task { try await peer.connectedSession(for: request) } + try await Self.waitUntil { await builder.callCount() == 1 } + await builder.release(call: 0) + try await Self.waitUntil { await winner.isClosedGateIsWaiting() } + let secondCaller = Task { try await peer.connectedSession(for: request) } + try await Self.waitUntil { await builder.callCount() == 2 } + await winner.releaseIsClosedGate() + _ = try await firstCaller.value + await builder.release(call: 1) + try await Self.waitUntil { await loser.closeGateIsWaiting() } + + // The redundant close is in flight; invalidation evicts the winner + // before that close settles. The second caller must not receive the + // stale winner capture. + await peer.invalidate() + await loser.releaseCloseGate() + try await Self.waitUntil { await builder.callCount() == 3 } + await builder.release(call: 2) + + let session = try await secondCaller.value + #expect(await session.connectionContinuityID() == 93) + #expect(await peer.connectionContinuityID() == 93) + #expect(await winner.closeCount() == 1) + #expect(await loser.closeCount() == 1) + await peer.invalidate() + } + + @Test + func invalidationDuringPostProbeRedundantDialCloseTriggersAFreshDial() async throws { + let request = try Self.request() + let peerID = try CmxConnectivityPeerID(request: request) + let winner = TestConnectivitySession( + continuityID: 101, + gatesFirstIsClosedCheck: true + ) + let loser = TestConnectivitySession( + continuityID: 102, + gatesFirstIsClosedCheck: true, + gatesFirstClose: true + ) + let replacement = TestConnectivitySession(continuityID: 103) + let builder = OrderedGatedConnectivitySessionBuilder( + sessions: [winner, loser, replacement] + ) + let peer = CmxConnectivityPeerSession( + peerID: peerID, + buildSession: { request in + try await builder.build(request) + } + ) + + // Park both callers at their dead-on-arrival probes so the winner + // installs while the second caller is past its post-resolve check. + let firstCaller = Task { try await peer.connectedSession(for: request) } + try await Self.waitUntil { await builder.callCount() == 1 } + await builder.release(call: 0) + try await Self.waitUntil { await winner.isClosedGateIsWaiting() } + let secondCaller = Task { try await peer.connectedSession(for: request) } + try await Self.waitUntil { await builder.callCount() == 2 } + await builder.release(call: 1) + try await Self.waitUntil { await loser.isClosedGateIsWaiting() } + await winner.releaseIsClosedGate() + _ = try await firstCaller.value + await loser.releaseIsClosedGate() + try await Self.waitUntil { await loser.closeGateIsWaiting() } + + await peer.invalidate() + await loser.releaseCloseGate() + try await Self.waitUntil { await builder.callCount() == 3 } + await builder.release(call: 2) + + let session = try await secondCaller.value + #expect(await session.connectionContinuityID() == 103) + #expect(await peer.connectionContinuityID() == 103) + #expect(await winner.closeCount() == 1) + #expect(await loser.closeCount() == 1) + await peer.invalidate() + } + @Test func deadOnArrivalSessionIsClosedAndRedialedOnce() async throws { let request = try Self.request() @@ -563,6 +668,9 @@ private actor TestConnectivitySession: CmxConnectivitySession { private var isClosedGatePending: Bool private var isClosedGateWaiting = false private var isClosedGateWaiter: CheckedContinuation? + private var closeGatePending: Bool + private var closeGateWaiting = false + private var closeGateWaiter: CheckedContinuation? private var received: [Data] = [] private var selectedPath = CmxIrohObservedConnectionPath.direct private var selectedPathContinuation: @@ -572,12 +680,14 @@ private actor TestConnectivitySession: CmxConnectivitySession { continuityID: UInt64, gatesCloseAttribution: Bool = false, keepsSelectedPathStreamOpen: Bool = false, - gatesFirstIsClosedCheck: Bool = false + gatesFirstIsClosedCheck: Bool = false, + gatesFirstClose: Bool = false ) { self.continuityID = continuityID self.gatesCloseAttribution = gatesCloseAttribution self.keepsSelectedPathStreamOpen = keepsSelectedPathStreamOpen isClosedGatePending = gatesFirstIsClosedCheck + closeGatePending = gatesFirstClose } func receiveControl(maximumByteCount: Int) -> Data? { @@ -681,11 +791,28 @@ private actor TestConnectivitySession: CmxConnectivitySession { } } - func close() { + func close() async { + if closeGatePending { + closeGatePending = false + closeGateWaiting = true + await withCheckedContinuation { continuation in + closeGateWaiter = continuation + } + closeGateWaiting = false + } closes += 1 finish(failure: .cancelled) } + func closeGateIsWaiting() -> Bool { + closeGateWaiting + } + + func releaseCloseGate() { + closeGateWaiter?.resume() + closeGateWaiter = nil + } + func finishRemotely(failure: DiagnosticFailureKind) { finish(failure: failure) }