diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift index 1012df1e21a4..a1287fed5d7d 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityEngine.swift +++ b/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? 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 + } } } diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift index c8e4aef5c044..fe90245f74a9 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityPeerSession.swift +++ b/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, diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxConnectivityRouteContent.swift new file mode 100644 index 000000000000..8bdcc4a90530 --- /dev/null +++ b/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] + } +} diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime+PolicyRefresh.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime+PolicyRefresh.swift index 71948caaaca6..965fa3098efc 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime+PolicyRefresh.swift +++ b/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 diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime.swift index ff5657c3b6a3..98b50c1f7d23 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohClientRuntime.swift +++ b/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 diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PolicyRefresh.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PolicyRefresh.swift index c93a8004615d..4d41d3eab5e5 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PolicyRefresh.swift +++ b/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, diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PublicAPI.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PublicAPI.swift index 288b93620777..75414a9ba1fd 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime+PublicAPI.swift +++ b/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, diff --git a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime.swift b/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime.swift index afede8b7bb29..ad1e950cf843 100644 --- a/Packages/Shared/CmuxIrohTransport/Sources/CmuxIrohTransport/CmxIrohHostRuntime.swift +++ b/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, diff --git a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift index 2ad489c000a9..e3a492d8f777 100644 --- a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift +++ b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityEngineTests.swift @@ -124,6 +124,168 @@ struct CmxConnectivityEngineTests { await engine.stop() } + @Test + func equivalentRouteRevisionBumpKeepsTheLivePeerSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z" + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + #expect(await session.isClosed() == false) + + try await rig.engine.reconcileRoutes() + + let snapshot = await rig.engine.snapshot() + #expect(snapshot.routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + #expect(snapshot.peers.first?.phase == .connected) + await rig.engine.stop() + } + + @Test + func changedIdentityGenerationOnRevisionBumpStillInvalidatesTheSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + identityGeneration: 2 + ), + ]) + 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() == 1) + #expect(await session.isClosed()) + await rig.engine.stop() + } + + @Test + func removedPeerBindingOnRevisionBumpStillInvalidatesTheSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + includesPeerBinding: false + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.connection.observedCloseCallCount() == 1) + #expect(await session.isClosed()) + await rig.engine.stop() + } + + @Test + func changedRelayFleetOnRevisionBumpStillInvalidatesTheSession() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z", + relayFleet: ["https://replacement.relay.example/"] + ), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.connection.observedCloseCallCount() == 1) + #expect(await session.isClosed()) + await rig.engine.stop() + } + + @Test + func revisionBumpWithoutReplacementContentFailsClosed() async throws { + let rig = try await Self.admittedPeerRig(responses: [ + Self.peerRouteResponse( + revision: 9, + lastSeenAt: "2026-07-30T00:00:00Z" + ), + Self.unchangedResponse(revision: 12), + ]) + let session = try await rig.engine.acquireControl( + for: rig.request, + ownerID: UUID() + ) + + try await rig.engine.reconcileRoutes() + + #expect(await rig.engine.snapshot().routeRevision == 12) + #expect(await rig.connection.observedCloseCallCount() == 1) + #expect(await session.isClosed()) + await rig.engine.stop() + } + + @Test + func installedRouteRevisionUsesRouteContentEquivalence() 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 equivalent = try #require(Self.peerRouteResponse( + revision: 10, + lastSeenAt: "2026-07-30T00:00:45Z" + ).snapshot) + let changed = try #require(Self.peerRouteResponse( + revision: 11, + lastSeenAt: "2026-07-30T00:01:30Z", + identityGeneration: 2 + ).snapshot) + + await rig.engine.didInstallRouteRevision(10, routes: equivalent) + + #expect(await rig.engine.snapshot().routeRevision == 10) + #expect(await rig.connection.observedCloseCallCount() == 0) + #expect(await session.isClosed() == false) + + await rig.engine.didInstallRouteRevision(11, routes: changed) + + #expect(await rig.engine.snapshot().routeRevision == 11) + #expect(await rig.connection.observedCloseCallCount() == 1) + #expect(await session.isClosed()) + await rig.engine.stop() + } + @Test func stopFinishesNetworkChangeObservers() async throws { let identity = try CmxIrohPeerIdentity( @@ -150,6 +312,125 @@ struct CmxConnectivityEngineTests { try await Self.waitUntil { await finished.value() } } + private static let peerEndpointID = String(repeating: "f", count: 64) + private static let peerDeviceID = "123e4567-e89b-42d3-a456-426614174999" + + private struct AdmittedPeerRig { + let engine: CmxConnectivityEngine + let connection: TestIrohConnection + let authority: ScriptedConnectivityAuthority + let request: CmxByteTransportRequest + } + + private static func admittedPeerRig( + responses: [CmxConnectivitySyncResponse] + ) 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 endpoint = TestDialingIrohEndpoint( + localIdentity: localIdentity, + dialResults: [.connection(connection)] + ) + let supervisor = CmxIrohEndpointSupervisor( + factory: TestIrohEndpointFactory(endpoints: [endpoint]), + configuration: try endpointConfiguration() + ) + let authority = ScriptedConnectivityAuthority(responses: responses) + let context = CmxIrohClientContext( + dialPlan: try testIrohDialPlan(), + credential: try .pairGrant("e30.e30.AA") + ) + let engine = CmxConnectivityEngine( + supervisor: supervisor, + contextProvider: TestIrohClientContextProvider(context: context), + authority: authority, + installRouteSnapshot: { _ in } + ) + try await engine.start() + let request = CmxByteTransportRequest( + route: try CmxAttachRoute( + id: "iroh-v2", + kind: .iroh, + endpoint: .peer(identity: peerIdentity, pathHints: []) + ), + expectedPeerDeviceID: peerDeviceID, + authorizationMode: .transportAdmission + ) + return AdmittedPeerRig( + engine: engine, + connection: connection, + authority: authority, + request: request + ) + } + + private static func peerRouteResponse( + revision: UInt64, + lastSeenAt: String, + identityGeneration: Int = 1, + relayFleet: [String] = ["https://relay.example/"], + includesPeerBinding: Bool = true + ) throws -> CmxConnectivitySyncResponse { + let binding = """ + { + "binding_id": "0a0a0a0a-0000-4000-8000-000000000001", + "device_id": "\(peerDeviceID)", + "app_instance_id": "0a0a0a0a-0000-4000-8000-000000000002", + "tag": "default", + "platform": "mac", + "endpoint_id": "\(peerEndpointID)", + "identity_generation": \(identityGeneration), + "pairing_enabled": true, + "capabilities": ["terminal"], + "path_hints": [], + "last_seen_at": "\(lastSeenAt)" + } + """ + let fleet = relayFleet + .map { "\"\($0)\"" } + .joined(separator: ", ") + return try decodeResponse( + """ + { + "protocol_version": 2, + "revision": \(revision), + "changed": true, + "reset": false, + "snapshot": { + "route_contract_version": 1, + "revision": \(revision), + "bindings": [\(includesPeerBinding ? binding : "")], + "relay_fleet": [\(fleet)], + "lan_rendezvous": { + "generation": 1, + "key": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + }, + "grant_verification_keys": { + "version": 1, + "current_kid": "current", + "keys": [] + } + } + } + """ + ) + } + private static func endpointConfiguration() throws -> CmxIrohEndpointConfiguration { CmxIrohEndpointConfiguration( secretKey: try CmxIrohSecretKey(bytes: Data(repeating: 5, count: 32)), @@ -224,6 +505,27 @@ struct CmxConnectivityEngineTests { } } +private actor ScriptedConnectivityAuthority: CmxConnectivityAuthorityServing { + private var responses: [CmxConnectivitySyncResponse] + private var observedKnownRevisions: [UInt64?] = [] + + init(responses: [CmxConnectivitySyncResponse]) { + self.responses = responses + } + + func syncConnectivity( + knownRevision: UInt64? + ) async throws -> CmxConnectivitySyncResponse { + observedKnownRevisions.append(knownRevision) + guard !responses.isEmpty else { + throw CmxIrohTrustBrokerClientError.connectivity + } + return responses.removeFirst() + } + + func knownRevisions() -> [UInt64?] { observedKnownRevisions } +} + private actor InitialThenFailingConnectivityAuthority: CmxConnectivityAuthorityServing { private let initial: CmxConnectivitySyncResponse private var calls = 0 diff --git a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift index ef1d0ed3e0c7..b39854ef2c84 100644 --- a/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift +++ b/Packages/Shared/CmuxIrohTransport/Tests/CmuxIrohTransportTests/CmxConnectivityPeerSessionTests.swift @@ -222,6 +222,50 @@ struct CmxConnectivityPeerSessionTests { await peer.releaseControl(ownerID: ownerID) } + @Test + func concurrentRedialCannotDisplaceAnInstalledLiveSession() async throws { + let request = try Self.request() + let peerID = try CmxConnectivityPeerID(request: request) + let winner = TestConnectivitySession( + continuityID: 81, + gatesFirstIsClosedCheck: true + ) + let loser = TestConnectivitySession( + continuityID: 82, + gatesFirstIsClosedCheck: true + ) + let builder = OrderedGatedConnectivitySessionBuilder( + sessions: [winner, loser] + ) + let peer = CmxConnectivityPeerSession( + peerID: peerID, + buildSession: { request in + try await builder.build(request) + } + ) + + // Park both callers past their pre-dial installed-slot checks so the + // first install lands while the second caller is still in flight. + 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 secondCaller.value + + #expect(await peer.connectionContinuityID() == 81) + #expect(await winner.closeCount() == 0) + #expect(await loser.closeCount() == 1) + #expect(await peer.snapshot().phase == .connected) + await peer.invalidate() + } + @Test func deadOnArrivalSessionIsClosedAndRedialedOnce() async throws { let request = try Self.request() @@ -516,6 +560,9 @@ private actor TestConnectivitySession: CmxConnectivitySession { private var closureWaiters: [CheckedContinuation] = [] private var closeAttributionWaiter: CheckedContinuation? private var closeAttributionWaiting = false + private var isClosedGatePending: Bool + private var isClosedGateWaiting = false + private var isClosedGateWaiter: CheckedContinuation? private var received: [Data] = [] private var selectedPath = CmxIrohObservedConnectionPath.direct private var selectedPathContinuation: @@ -524,11 +571,13 @@ private actor TestConnectivitySession: CmxConnectivitySession { init( continuityID: UInt64, gatesCloseAttribution: Bool = false, - keepsSelectedPathStreamOpen: Bool = false + keepsSelectedPathStreamOpen: Bool = false, + gatesFirstIsClosedCheck: Bool = false ) { self.continuityID = continuityID self.gatesCloseAttribution = gatesCloseAttribution self.keepsSelectedPathStreamOpen = keepsSelectedPathStreamOpen + isClosedGatePending = gatesFirstIsClosedCheck } func receiveControl(maximumByteCount: Int) -> Data? { @@ -575,7 +624,26 @@ private actor TestConnectivitySession: CmxConnectivitySession { ) } - func isClosed() -> Bool { closed } + func isClosed() async -> Bool { + if isClosedGatePending { + isClosedGatePending = false + isClosedGateWaiting = true + await withCheckedContinuation { continuation in + isClosedGateWaiter = continuation + } + isClosedGateWaiting = false + } + return closed + } + + func isClosedGateIsWaiting() -> Bool { + isClosedGateWaiting + } + + func releaseIsClosedGate() { + isClosedGateWaiter?.resume() + isClosedGateWaiter = nil + } func connectionContinuityID() -> UInt64? { closed ? nil : continuityID