diff --git a/Packages/Shared/CMUXMobileCore/Sources/CMUXMobileCore/MobileSyncProtocol.swift b/Packages/Shared/CMUXMobileCore/Sources/CMUXMobileCore/MobileSyncProtocol.swift index e23e26d95bd0..63e0f58456b6 100644 --- a/Packages/Shared/CMUXMobileCore/Sources/CMUXMobileCore/MobileSyncProtocol.swift +++ b/Packages/Shared/CMUXMobileCore/Sources/CMUXMobileCore/MobileSyncProtocol.swift @@ -201,6 +201,10 @@ public struct MobileSyncFrameCodec { return frame } + /// Decodes at most one bounded batch, leaving later frames in `buffer`. + /// Callers drain another batch before awaiting more network bytes whenever + /// the returned count reaches `maximumDecodedFrameCount`. A transport read + /// may coalesce any number of valid frames; its boundary is not a message limit. public static func decodeFrames( from buffer: inout Data, maximumFrameByteCount: Int = defaultMaximumFrameByteCount, @@ -222,7 +226,8 @@ public struct MobileSyncFrameCodec { } } - while buffer.count - consumedByteCount >= headerByteCount { + while frames.count < maximumDecodedFrameCount, + buffer.count - consumedByteCount >= headerByteCount { let frameStart = buffer.index( buffer.startIndex, offsetBy: consumedByteCount @@ -241,11 +246,6 @@ public struct MobileSyncFrameCodec { guard buffer.count - consumedByteCount >= headerByteCount + payloadLength else { break } - guard frames.count < maximumDecodedFrameCount else { - throw MobileSyncFrameCodecError.tooManyFrames( - maximumDecodedFrameCount - ) - } let payloadStart = headerEnd let payloadEnd = buffer.index( payloadStart, diff --git a/Packages/Shared/CMUXMobileCore/Tests/CMUXMobileCoreTests/MobileSyncProtocolTests.swift b/Packages/Shared/CMUXMobileCore/Tests/CMUXMobileCoreTests/MobileSyncProtocolTests.swift index 6c97dde42115..d6a08205ee8e 100644 --- a/Packages/Shared/CMUXMobileCore/Tests/CMUXMobileCoreTests/MobileSyncProtocolTests.swift +++ b/Packages/Shared/CMUXMobileCore/Tests/CMUXMobileCoreTests/MobileSyncProtocolTests.swift @@ -199,39 +199,31 @@ import Testing } } -@Test func frameCodecRejectsAZeroLengthFrameFloodAtTheCallerLimit() throws { - let emptyFrame = try MobileSyncFrameCodec.encodeFrame(Data()) +@Test func frameCodecDrainsLargeBatchesWithoutTreatingPacketSizeAsFailure() throws { + let frame = try MobileSyncFrameCodec.encodeFrame(Data("valid payload".utf8)) var buffer = Data() - for _ in 0..<10_000 { - buffer.append(emptyFrame) - } - - do { - _ = try MobileSyncFrameCodec.decodeFrames( - from: &buffer, - maximumDecodedFrameCount: 16 - ) - Issue.record("Expected the decoded frame count limit to fail closed") - } catch let error as MobileSyncFrameCodecError { - #expect(error == .tooManyFrames(16)) - #expect(buffer.count == (10_000 - 16) * emptyFrame.count) + for _ in 0..<1_000 { buffer.append(frame) } + var decoded = 0 + while !buffer.isEmpty { + let frames = try MobileSyncFrameCodec.decodeFrames(from: &buffer, maximumDecodedFrameCount: 16) + #expect(!frames.isEmpty) + #expect(frames.count <= 16) + #expect(frames.allSatisfy { $0 == Data("valid payload".utf8) }) + decoded += frames.count } + #expect(decoded == 1_000) } @Test func frameCodecDefaultFrameCountLimitIsFinite() throws { - let emptyFrame = try MobileSyncFrameCodec.encodeFrame(Data()) + let frame = try MobileSyncFrameCodec.encodeFrame(Data()) var buffer = Data() - for _ in 0...MobileSyncFrameCodec.defaultMaximumDecodedFrameCount { - buffer.append(emptyFrame) - } - - #expect(throws: MobileSyncFrameCodecError.tooManyFrames( - MobileSyncFrameCodec.defaultMaximumDecodedFrameCount - )) { - _ = try MobileSyncFrameCodec.decodeFrames(from: &buffer) - } + for _ in 0...MobileSyncFrameCodec.defaultMaximumDecodedFrameCount { buffer.append(frame) } + let frames = try MobileSyncFrameCodec.decodeFrames(from: &buffer) + #expect(frames.count == MobileSyncFrameCodec.defaultMaximumDecodedFrameCount) + #expect(buffer == frame) } + private func base64URLEncode(_ data: Data) -> String { data.base64EncodedString() .replacingOccurrences(of: "+", with: "-") diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxConnection.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxConnection.swift index e0a311f25f83..953ad1e6ff97 100644 --- a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxConnection.swift +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxConnection.swift @@ -178,7 +178,7 @@ public actor IrxConnection { } /// Returns recent diagnostic evidence. A false result does not establish - /// peer death; callers must probe before replacing an open connection. + /// peer death; only native connection closure establishes transport failure. public func hasRecentKeepalive(within age: Duration) -> Bool { guard let lastPongAt else { return false } return ContinuousClock.now - lastPongAt <= age @@ -261,42 +261,50 @@ public actor IrxConnection { return writer } - /// Accepts the next bidirectional lane (server side, post-admission). - /// Returns nil once the connection is closed. + /// Accepts the next bidirectional lane. A malformed descriptor retires + /// only that stream; native connection termination ends acceptance. public func acceptLane() async -> IrxLaneStream? { - while true { + while !Task.isCancelled { do { let stream = try await connection.acceptBi() let reader = IrxStreamReader(stream.recv()) let writer = IrxStreamWriter(stream.send()) - guard - let descriptor = try await reader.readControlFrame( - IrxLaneDescriptor.self) - else { - await writer.finish() - continue + do { + if let descriptor = try await reader.readControlFrame(IrxLaneDescriptor.self) { + return IrxLaneStream(descriptor: descriptor, writer: writer, reader: reader) + } + } catch { + // Stream framing failure does not change native connection state. } - return IrxLaneStream(descriptor: descriptor, writer: writer, reader: reader) + await writer.reset(errorCode: 2) + await reader.stop(errorCode: 2) } catch { - closedFlag = true return nil } } + return nil } - /// Accepts the next unidirectional lane (client side: the events lane). + /// Accepts the next usable server event lane without treating a malformed + /// descriptor or stream EOF as complete-connection closure. public func acceptUniLane() async throws -> (IrxLaneDescriptor, IrxStreamReader)? { - do { - let stream = try await connection.acceptUni() - let reader = IrxStreamReader(stream) - guard - let descriptor = try await reader.readControlFrame(IrxLaneDescriptor.self) - else { return nil } - return (descriptor, reader) - } catch { - closedFlag = true - return nil + while !Task.isCancelled { + do { + let stream = try await connection.acceptUni() + let reader = IrxStreamReader(stream) + do { + if let descriptor = try await reader.readControlFrame(IrxLaneDescriptor.self) { + return (descriptor, reader) + } + } catch { + // A later event stream may repair this optional feature. + } + await reader.stop(errorCode: 2) + } catch { + return nil + } } + return nil } /// The selected QUIC path right now, for relay attribution evidence. @@ -308,8 +316,9 @@ public actor IrxConnection { return "\(selected.isRelay ? "relay" : "direct"):\(selected.remoteAddr)" } - /// Starts one periodic peer probe owner. A failed probe retires its stream; - /// another attempt uses a fresh stream on the same QUIC connection. + /// Samples application round-trip latency on an optional lane. + /// A failed probe retires its stream; another attempt uses a fresh stream + /// on the same QUIC connection. Native closure owns dead-peer detection. public func startClientKeepalive( interval: Duration = IrxProtocol.keepaliveInterval, deadline: Duration = IrxProtocol.keepaliveDeadline, @@ -321,7 +330,7 @@ public actor IrxConnection { } /// Pauses application probes before suspension without closing QUIC. - /// A resumed loop begins with zero strikes and a fresh probe deadline. + /// A resumed loop starts with a fresh probe stream and deadline. public func setApplicationActive(_ active: Bool) { guard applicationActive != active else { return } applicationActive = active @@ -336,7 +345,7 @@ public actor IrxConnection { /// Tests the existing peer with a bounded ping/pong exchange. Concurrent /// callers share the current probe; a shorter caller deadline also retires /// that probe, so no caller retries a stopped receive stream. - /// A false result during suspension is inconclusive and must not cause a close. + /// A false result is inconclusive and never authorizes connection teardown. public func probeLiveness(deadline: Duration = IrxProtocol.keepaliveDeadline) async -> Bool { guard applicationActive, !isClosed, !Task.isCancelled else { return false } if let task = probeTask, let id = probeID { @@ -365,25 +374,19 @@ public actor IrxConnection { keepaliveGeneration &+= 1 let generation = keepaliveGeneration keepaliveTask = Task { - var strikes = 0 while !Task.isCancelled, self.applicationActive, self.keepaliveGeneration == generation { - if strikes == 0 { - do { try await Task.sleep(for: settings.interval) } catch { return } - } + do { try await Task.sleep(for: settings.interval) } catch { return } guard !Task.isCancelled, self.applicationActive, self.keepaliveGeneration == generation else { return } let alive = await self.probeLiveness(deadline: settings.deadline) guard !Task.isCancelled, self.applicationActive, self.keepaliveGeneration == generation else { return } - if alive { strikes = 0; continue } - strikes += 1 - if strikes < IrxProtocol.keepaliveStrikeLimit { - self.journal.record("keepalive", "miss", ["strike": String(strikes), - "path": self.selectedPathDescription()]) - continue + if alive { continue } + self.journal.record("keepalive", "miss", ["path": self.selectedPathDescription()]) + // Probe silence retires only its diagnostic stream. The native + // watcher continues to observe closure while fresh probes retry. + if self.isClosed { + await settings.onDeath() + return } - self.journal.record("keepalive", "timeout", ["path": self.selectedPathDescription()]) - await self.close(code: .keepaliveTimeout, origin: .transport) - await settings.onDeath() - return } } } diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxControlByteTransport.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxControlByteTransport.swift index 7187294a4a54..4f80abcaec6e 100644 --- a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxControlByteTransport.swift +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxControlByteTransport.swift @@ -11,6 +11,9 @@ public import Foundation /// session, so the transport retires that exact session before closing the /// complete QUIC connection. Session ownership belongs to /// ``IrxPeerEngine``. +/// +/// Each transport binds to at most one admitted session. Native closure makes +/// that transport terminal; the RPC owner creates a new transport to recover. public actor IrxControlByteTransport: CmxByteTransport { /// Factory for an admitted connection and its control lane. public typealias Establish = @Sendable () async throws -> (IrxConnection, IrxLaneStream) @@ -113,7 +116,15 @@ public actor IrxControlByteTransport: CmxByteTransport { private func establishedPair() async throws -> (IrxConnection, IrxLaneStream) { guard !isClosed else { throw IrxConnectionError.closed(nil) } - if let pair, await !pair.0.isConnectionClosed() { + if let pair { + let connectionIsClosed = await pair.0.isConnectionClosed() + guard !isClosed else { throw IrxConnectionError.closed(nil) } + if connectionIsClosed { + // Reads and writes may still be unwinding on this pair. Keep + // their eventual close tied to this RPC generation's session. + await close() + throw IrxConnectionError.closed(nil) + } return pair } if let connectInFlight { diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxPeerEngine.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxPeerEngine.swift index d33bd1f2863d..c296bc1b2477 100644 --- a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxPeerEngine.swift +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxPeerEngine.swift @@ -36,7 +36,7 @@ public struct IrxClientSession: Sendable { } /// THE single reconnect owner for one Mac peer (iOS side). Every trigger - -/// app recovery, foreground, network change, keepalive death, explicit retry - +/// app recovery, foreground, network change, native closure, explicit retry - /// is an input; automatic triggers JOIN the in-flight dial, explicit intent /// replaces it. Transport failures retry on a capped backoff that resets on /// success; denials and supersession park the engine until an explicit @@ -295,9 +295,9 @@ public actor IrxPeerEngine { if hasConnectionIntent { foregroundKick() } } - /// Retains an open session only after a fresh bounded liveness check. Each - /// unsuccessful attempt uses a live stream. Known native closure bypasses - /// probing; age of the last pong never forces replacement. + /// Samples the existing session on a fresh probe deadline after suspension. + /// An unanswered probe leaves the native connection in place; application + /// owners repair their streams without replacing the admitted session. public func foregroundKick() { guard applicationActive, foregroundTask == nil, hasConnectionIntent, parkedCode == nil else { return } let generation = activityGeneration @@ -310,19 +310,26 @@ public actor IrxPeerEngine { private func recoverOnForeground(generation: UInt64) async { let started = clockNow() if let previous = session { - for _ in 0.. [(IrxConnection, IrxLaneStream)] in + var pairs: [(IrxConnection, IrxLaneStream)] = [] + for _ in 0..<2 { + guard let incoming = await server.acceptNext() else { break } + let native = try await incoming.accept().connect() + let connection = IrxConnection( + connection: native, role: .acceptor, journal: journal) + guard let (_, control, _) = await IrxAdmission.performServer( + connection: connection, + judgment: IrxLiveTestSupport.fixedJudgment(accepting: "good-grant"), + journal: journal + ) else { break } + pairs.append((connection, control)) + } + return pairs + } + var clientPairs: [(IrxConnection, IrxLaneStream)] = [] + for _ in 0..<2 { + let native = try await client.connect( + addr: IrxLiveTestSupport.loopbackAddr(of: server), alpn: IrxProtocol.alpnData) + let connection = IrxConnection(connection: native, role: .dialer, journal: journal) + let (_, control) = try await IrxAdmission.performClient( + connection: connection, grantJWS: "good-grant", journal: journal) + clientPairs.append((connection, control)) + } + let serverPairs = try await serverTask.value + try #require(serverPairs.count == 2) + let first = clientPairs[0] + let replacement = clientPairs[1] + let establishments = AsyncStream.makeStream() + let releaseProbe = IrxControlReleaseProbe() + let transport = IrxControlByteTransport( + closeCode: .explicitRedial, + establish: { + establishments.continuation.yield(()) + return await first.0.isConnectionClosed() ? replacement : first + }, + onClose: { connection, code, retiresConnection in + #expect(connection.underlying.stableId() == first.0.underlying.stableId()) + await releaseProbe.record(closeCode: code, retiresConnection: retiresConnection) + } + ) + try await transport.connect() + try await transport.connect() + await serverPairs[0].0.close(code: .hostShutdown, origin: .local) + // Synchronize with Iroh's native closure without receiving on the old + // control lane, which would independently mark the transport closed. + _ = await first.0.underlying.closed() + #expect(await transport.isTransportClosed()) + + do { + if sendAfterClosure { + try await transport.send(Data("old RPC generation".utf8)) + } else { + try await transport.connect() + } + Issue.record("closed control transport adopted a replacement session") + } catch let error as IrxConnectionError { + guard case .closed = error else { + Issue.record("unexpected connection error after native closure: \(error)") + return + } + } catch { + Issue.record("unexpected error after native closure: \(error)") + } + + // A delayed close from this RPC generation must only release its + // original claim; it must never reach the replacement connection. + await transport.close() + #expect(await releaseProbe.count == 1) + #expect(await releaseProbe.retiresConnections == [false]) + establishments.continuation.finish() + var establishmentCount = 0 + for await _ in establishments.stream { establishmentCount += 1 } + #expect(establishmentCount == 1) + #expect(await !replacement.0.isConnectionClosed()) + + let replacementTransport = IrxControlByteTransport( + connection: replacement.0, control: replacement.1, closeCode: .explicitRedial) + try await replacementTransport.connect() + let message = Data("new RPC generation".utf8) + let received = try await withIrxDeadline(.seconds(1), onTimeout: { + await serverPairs[1].1.reader.stop() + }) { + try await replacementTransport.send(message) + return try await serverPairs[1].1.reader.readRaw() + } + #expect(received == message) + await replacementTransport.close() + } + @Test("cancelling a control read retires the owner locally") func cancelledControlReadRetiresLocally() async throws { let journal = IrxLiveTestSupport.journal() diff --git a/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxLivenessTests.swift b/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxLivenessTests.swift index c0fb8cc3fc58..f0f0ec387ff6 100644 --- a/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxLivenessTests.swift +++ b/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxLivenessTests.swift @@ -32,7 +32,23 @@ struct IrxLivenessTests { await session.connection.close(code: .userRequested, origin: .local) } - @Test func suspensionDuringAProbeCannotAccumulateStrikes() async throws { + @Test func repeatedProbeTimeoutsPreserveNativeConnectionAndControlTraffic() async throws { + let host = try await IrxLivenessTestHost.make(behavior: .ignoreFirstConnection) + defer { Task { await host.stop() } } + let session = try await host.dial() + try await session.connection.startClientKeepalive(interval: .milliseconds(10), deadline: .milliseconds(100)) { + await host.recordDeath() + } + try await waitUntil { host.journal.counterSnapshot()["miss", default: 0] >= 3 } + #expect(await host.probeCount >= 3) + #expect(await host.connectionCount == 1) + #expect(await host.deathCount == 0) + #expect(await !session.connection.isClosed) + try await expectControlRoundTrip(on: session, message: "control-survives-probe-timeouts") + await session.connection.close(code: .userRequested, origin: .local) + } + + @Test func suspensionCancelsProbeWithoutDeclaringFailure() async throws { let host = try await IrxLivenessTestHost.make(behavior: .delayFirstProbe) defer { Task { await host.stop() } } let session = try await host.dial() @@ -41,7 +57,7 @@ struct IrxLivenessTests { } try await waitUntil { await host.probeCount == 1 } await session.connection.setApplicationActive(false) - // Deliberately outlast both old probe deadlines. This represents time + // Deliberately outlast the cancelled probe deadline. This represents time // during which iOS is backgrounded and cannot perform application work. try await Task.sleep(for: .milliseconds(250)) #expect(host.journal.counterSnapshot()["miss", default: 0] == 0) @@ -101,7 +117,7 @@ struct IrxLivenessTests { await engine.stop() } - @Test func silentPeerGetsBoundedFreshProbesBeforeForegroundReplacement() async throws { + @Test func unansweredForegroundProbeRetainsSessionAndControlTraffic() async throws { let host = try await IrxLivenessTestHost.make(behavior: .ignoreFirstConnection) defer { Task { await host.stop() } } let engine = IrxPeerEngine(config: .init(keepaliveInterval: .seconds(60)), journal: host.journal) { @@ -109,21 +125,47 @@ struct IrxLivenessTests { } let first = try await engine.ensureSession(trigger: "test") await engine.setApplicationActive(false) - let started = ContinuousClock.now await engine.setApplicationActive(true) - try await waitUntil { - guard let current = await engine.currentSession() else { return false } - return current.admit.session != first.admit.session + try await waitUntil { host.journal.counterSnapshot()["foreground-session-retained", default: 0] > 0 } + #expect(await engine.currentSession()?.admit.session == first.admit.session) + #expect(await host.probeCount == 1) + #expect(await host.connectionCount == 1) + #expect(await !first.connection.isClosed) + #expect(host.journal.counterSnapshot()["session-ended", default: 0] == 0) + #expect(host.journal.counterSnapshot()["auto-redial", default: 0] == 0) + try await expectControlRoundTrip(on: first, message: "control-survives-foreground-probe") + await engine.stop() + } + + @Test func terminalPeerCloseRemainsParkedAcrossForeground() async throws { + let host = try await IrxLivenessTestHost.make(behavior: .respond) + defer { Task { await host.stop() } } + let engine = IrxPeerEngine(config: .init(keepaliveInterval: .seconds(60)), journal: host.journal) { + try await host.dial() } - let elapsed = started.duration(to: .now) - #expect(elapsed < .seconds(2)) - host.journal.record("acceptance", "silent-peer-resume", ["elapsed": String(describing: elapsed)]) - #expect(await host.probeCount == 2) - #expect(await host.connectionCount == 2) - #expect(await first.connection.isClosed) + _ = try await engine.ensureSession(trigger: "test") + await engine.setApplicationActive(false) + await host.closeFirstConnection(code: .superseded) + try await waitUntil { host.journal.counterSnapshot()["auto-redial-suppressed", default: 0] > 0 } + await engine.setApplicationActive(true) + await engine.foregroundKick() + #expect(await engine.currentState == .closed(code: IrxCloseCode.superseded.rawValue)) + #expect(await engine.currentSession() == nil) + #expect(await host.connectionCount == 1) + #expect(host.journal.counterSnapshot()["auto-redial", default: 0] == 0) await engine.stop() } + private func expectControlRoundTrip(on session: IrxClientSession, message: String) async throws { + let response = try await withIrxDeadline(.seconds(1), onTimeout: { + await session.control.reader.stop() + }) { + try await session.control.writer.writeControlFrame(message) + return try await session.control.reader.readControlFrame(String.self) + } + #expect(response == message) + } + @Test func deferredWarmupResumesWithoutAUserAction() async throws { let host = try await IrxLivenessTestHost.make(behavior: .respond) defer { Task { await host.stop() } } @@ -174,10 +216,17 @@ private actor IrxLivenessTestHost { do { let native = try await incoming.accept().connect() let connection = IrxConnection(connection: native, role: .acceptor, journal: journal) - guard await IrxAdmission.performServer(connection: connection, - judgment: IrxLiveTestSupport.fixedJudgment(accepting: "good-grant"), journal: journal) != nil else { continue } + guard let (_, control, _) = await IrxAdmission.performServer(connection: connection, + judgment: IrxLiveTestSupport.fixedJudgment(accepting: "good-grant"), journal: journal) else { continue } connections.append(connection) let index = connections.count + tasks.append(Task { + do { + while let message = try await control.reader.readControlFrame(String.self), !Task.isCancelled { + try await control.writer.writeControlFrame(message) + } + } catch { /* Closing the test connection ends its echo stream. */ } + }) tasks.append(Task { while let lane = await connection.acceptLane(), !Task.isCancelled { guard lane.descriptor.lane == .keepalive else { continue } @@ -217,7 +266,9 @@ private actor IrxLivenessTestHost { } func recordDeath() { deathCount += 1 } - func closeFirstConnection() async { await connections.first?.close(code: .hostShutdown, origin: .local) } + func closeFirstConnection(code: IrxCloseCode = .hostShutdown) async { + await connections.first?.close(code: code, origin: .local) + } func stop() async { await releaseResponse.signal() tasks.forEach { $0.cancel() } diff --git a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession+IndependentEvents.swift b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession+IndependentEvents.swift index baff29f16f77..cc92e2fa1c0e 100644 --- a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession+IndependentEvents.swift +++ b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession+IndependentEvents.swift @@ -92,15 +92,18 @@ extension MobileCoreRPCSession { for try await chunk in stream { try Task.checkCancellation() guard !chunk.isEmpty else { continue } - guard chunk.count <= Self.maximumReceiveBufferByteCount - buffer.count else { - throw MobileSyncFrameCodecError.frameTooLarge(buffer.count + chunk.count) - } + // The codec checks each frame's size, independent of how the + // transport groups adjacent frames into chunks. buffer.append(chunk) - let frames = try MobileSyncFrameCodec.decodeFrames( - from: &buffer, - maximumDecodedFrameCount: Self.maximumDecodedFrameCountPerRead - ) - for frame in frames { dispatch(frame: frame) } + while !Task.isCancelled, independentEventReader?.id == id { + let frames = try MobileSyncFrameCodec.decodeFrames( + from: &buffer, + maximumDecodedFrameCount: Self.maximumDecodedFrameCountPerRead + ) + for frame in frames { dispatch(frame: frame) } + guard frames.count == Self.maximumDecodedFrameCountPerRead else { break } + await Task.yield() + } } } catch { // The host falls back to control delivery after optional-lane failure. diff --git a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession.swift b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession.swift index ba62717db14d..87dafa941568 100644 --- a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession.swift +++ b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCSession.swift @@ -30,9 +30,6 @@ actor MobileCoreRPCSession { static let defaultAbandonedConnectCleanupTimeoutNanoseconds: UInt64 = 1_000_000_000 static let defaultLateAbandonedConnectCloseTimeoutNanoseconds: UInt64 = 5_000_000_000 static let defaultCancelledWriteCompletionGraceNanoseconds: UInt64 = 250_000_000 - static let maximumReceiveBufferByteCount = - MobileSyncFrameCodec.defaultMaximumFrameByteCount - + MobileSyncFrameCodec.headerByteCount static let maximumDecodedFrameCountPerRead = 256 struct EventSubscription { @@ -1039,30 +1036,23 @@ actor MobileCoreRPCSession { installedConnectionID == connectionID else { return } - guard chunk.count <= Self.maximumReceiveBufferByteCount - buffer.count else { - await tearDownIfInstalled( - connectionID: connectionID, - error: .invalidResponse - ) - return - } + // Enforce size per decoded frame. A chunk can finish one valid + // maximum-size frame and also contain bytes from the next frame. buffer.append(chunk) - let frames: [Data] do { - frames = try MobileSyncFrameCodec.decodeFrames( - from: &buffer, - maximumDecodedFrameCount: Self.maximumDecodedFrameCountPerRead - ) + while !Task.isCancelled, installedConnectionID == connectionID { + let frames = try MobileSyncFrameCodec.decodeFrames( + from: &buffer, + maximumDecodedFrameCount: Self.maximumDecodedFrameCountPerRead + ) + for frame in frames { dispatch(frame: frame) } + guard frames.count == Self.maximumDecodedFrameCountPerRead else { break } + await Task.yield() + } } catch { - await tearDownIfInstalled( - connectionID: connectionID, - error: .invalidResponse - ) + await tearDownIfInstalled(connectionID: connectionID, error: .invalidResponse) return } - for frame in frames { - dispatch(frame: frame) - } } } @@ -1235,7 +1225,10 @@ actor MobileCoreRPCSession { /// the wedged transport installed (their timeout cannot recycle a write /// owned by another request ID). private func startQueuedDemandRecovery(requestID: String) { - guard !queuedWriteIDs.isEmpty else { return } + // Native connection observation owns Iroh's lifetime. Queued demand + // has its own request deadline and cannot condemn an unfinished frame. + guard !(transport is any CmxByteTransportLivenessObserving), + !queuedWriteIDs.isEmpty else { return } Task { [self, taskTimeout, cancelledWriteCompletionGraceNanoseconds] in let waitTask = Task { await self.awaitCancelledWriteResolution() @@ -1266,6 +1259,10 @@ actor MobileCoreRPCSession { private func waitForCancelledActiveWriteResolution( deadlineUptimeNanoseconds: UInt64 ) async throws { + // Preserve serialization until the native write completes or fails. + // Cancelling writeAll can leave a frame prefix on the control stream. + // Later requests may queue and expire without cancelling that write. + if transport is any CmxByteTransportLivenessObserving { return } while let write = activeWrite, write.cancelledRequestResolutionTask != nil { try Task.checkCancellation() @@ -1367,7 +1364,12 @@ actor MobileCoreRPCSession { } private func recycleTransportIfActiveWrite(requestID: String) async -> Bool { - guard activeWrite?.requestID == requestID else { return false } + guard let write = activeWrite, write.requestID == requestID else { return false } + if let observing = transport as? any CmxByteTransportLivenessObserving { + guard await observing.isTransportClosed() else { return false } + guard activeWrite?.connectionID == write.connectionID, + activeWrite?.requestID == requestID else { return false } + } activeWrite?.task.cancel() activeWrite?.cancelledRequestResolutionTask?.cancel() activeWrite = nil diff --git a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/ControllableResponseTransport.swift b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/ControllableResponseTransport.swift index 8d0b751d8e9c..7901b4dad44e 100644 --- a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/ControllableResponseTransport.swift +++ b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/ControllableResponseTransport.swift @@ -95,9 +95,33 @@ actor ControllableResponseTransport: CmxByteTransport { } } + func deliverCoalescedResponse(id: String, eventCount: Int) throws { + let event = try MobileSyncFrameCodec.encodeFrame( + Data(#"{"kind":"event","topic":"workspace.updated","payload":{}}"#.utf8) + ) + var batch = Data() + for _ in 0.. Data { + try MobileCoreRPCClient.requestData(method: "terminal.input", params: ["text": id], id: id) + } +} diff --git a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/NativeObservedResponseTransport.swift b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/NativeObservedResponseTransport.swift new file mode 100644 index 000000000000..703fff6396b1 --- /dev/null +++ b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/NativeObservedResponseTransport.swift @@ -0,0 +1,13 @@ +import CMUXMobileCore +import Foundation + +/// Supplies the same independent connection-state snapshot as the Iroh adapter. +struct NativeObservedResponseTransport: CmxByteTransport, CmxByteTransportLivenessObserving { + let base: ControllableResponseTransport + + func connect() async throws { try await base.connect() } + func send(_ data: Data) async throws { try await base.send(data) } + func receive() async throws -> Data? { try await base.receive() } + func close() async { await base.close() } + func isTransportClosed() async -> Bool { await base.closed() } +} diff --git a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/DeviceRegistryService.swift b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/DeviceRegistryService.swift index 06c6120abeb0..d54a8823eca8 100644 --- a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/DeviceRegistryService.swift +++ b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/DeviceRegistryService.swift @@ -813,14 +813,22 @@ public extension MobileIOSAppNamespace { deviceWitness: String? = nil, evidence: any SameDeviceEvidenceProbing = IrohEndpointIdentityEvidenceProbe() ) -> String? { - DeviceRegistryService.durableDeviceID( - store: KeychainDeviceIdentityStore( - service: keychainService( - base: "com.cmuxterm.deviceRegistry.iosDeviceID.v1" - ), - accessGroup: keychainAccessGroup, - legacyService: "com.cmuxterm.deviceRegistry.iosDeviceID.v1" + #if targetEnvironment(simulator) + let store: any DeviceIdentityStoring = SimulatorDeviceIdentityStore( + defaults: defaults, + seededDeviceID: ProcessInfo.processInfo.environment["CMUX_SIMULATOR_DEVICE_ID"] + ) + #else + let store: any DeviceIdentityStoring = KeychainDeviceIdentityStore( + service: keychainService( + base: "com.cmuxterm.deviceRegistry.iosDeviceID.v1" ), + accessGroup: keychainAccessGroup, + legacyService: "com.cmuxterm.deviceRegistry.iosDeviceID.v1" + ) + #endif + return DeviceRegistryService.durableDeviceID( + store: store, defaults: defaults, deviceWitness: deviceWitness, evidence: evidence diff --git a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite+ConnectionRecovery.swift b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite+ConnectionRecovery.swift index 605ded741bcb..4f6b78734b0e 100644 --- a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite+ConnectionRecovery.swift +++ b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite+ConnectionRecovery.swift @@ -148,14 +148,40 @@ extension MobileShellComposite { } } - /// A definitive event-stream failure bypasses same-client resubscription. - /// Once the exact session is proven dead, rebuilding its listener only hides - /// the failure behind the transport's reconnect behavior and leaves the - /// shell owner stale. Instead, transition the one lifecycle owner to a fresh - /// authenticated stored-Mac dial. + /// Checks native connection state before promoting a feature failure to + /// recovery. An event stream can end while Iroh has already replaced the + /// underlying path, so the stream ending alone does not justify retiring + /// the RPC client. Transports without native observation retain their + /// existing error-driven recovery; Iroh owns its own dead-peer detection. func recoverDeadConnection( trigger: RecoveryTrigger, expectedClient: MobileCoreRPCClient + ) { + guard remoteClient === expectedClient, connectionState == .connected else { return } + Task { @MainActor [weak self] in + let closed = await expectedClient.isTransportClosed() + guard let self, + self.remoteClient === expectedClient, + self.connectionState == .connected else { return } + if closed == false { + // A failed subscription or request is feature-level evidence. + // Native Iroh closure is observed independently by the RPC session. + if trigger == .subscriptionStartFailed || trigger == .eventStreamEnded { + self.resyncTerminalOutput( + reason: "event_subscription_repair", + restartEventStream: true, + recoversConnectionOnSubscriptionFailure: false + ) + } + return + } + self.recoverClosedControlSession(trigger: trigger, expectedClient: expectedClient) + } + } + + func recoverClosedControlSession( + trigger: RecoveryTrigger, + expectedClient: MobileCoreRPCClient ) { guard remoteClient === expectedClient, connectionState == .connected else { return } guard foregroundRefreshIsActive else { @@ -233,7 +259,6 @@ extension MobileShellComposite { // cannot safely invent a redial route and must remain unavailable. switch trigger { case .liveness, .networkChange: - markMacConnectionReconnecting() resyncTerminalOutput(reason: trigger.description, restartEventStream: true) case .manual, .presencePush, .directoryChanged, .foreground, .eventStreamEnded, .subscriptionStartFailed, .transportWriteTimedOut, .automaticBackoffExpired, @@ -288,6 +313,30 @@ extension MobileShellComposite { self.applyConnectionRecoveryOwnerState() return } + let transportClosed = await expectedClient.isTransportClosed() + guard !Task.isCancelled, + self.connectionRecoveryOwner.isCurrent(attempt), + self.remoteClient === expectedClient, + self.connectionGeneration == attempt.sourceConnectionGeneration else { return } + if transportClosed == false { + // A slow application response after a path change or + // resume does not invalidate the native connection. + _ = self.completeConnectionRecovery(attempt) + self.markMacConnectionHealthy() + // The probe timed out, so the control request does + // not prove that the terminal event stream survived + // the background transition. Keep the native session + // and repair only the feature lane that can leave a + // mounted Ghostty surface blank. + if resyncAfterHealthy { + self.resyncTerminalOutput( + reason: "connectionRecovery.\(trigger).transportAlive", + restartEventStream: true + ) + } + self.applyConnectionRecoveryOwnerState() + return + } if self.lastBackgroundedAt != nil || self.foregroundResumeEpoch != epochAtProbeStart { // The probe spanned a background window: its wall-clock @@ -313,14 +362,7 @@ extension MobileShellComposite { self.connectionRecoveryOwner.transitionToRedialing(attempt) else { return } if let expectedClient { guard self.remoteClient === expectedClient else { return } - // Detach the stale shell synchronously on the main actor - // before awaiting its transport teardown. This cancels every - // tracked producer and makes untracked producers fail their - // identity guard, so they cannot reopen the old endpoint - // while the fresh stored-Mac dial starts. - self.connectionState = .disconnected - self.macConnectionStatus = .unavailable - self.clearRemoteConnectionContext() + self.retireRemoteClientForConnectionRecovery() self.applyConnectionRecoveryOwnerState() MobileDebugLog.anchormux( "connection.recovery waiting for physical transport drain " @@ -348,11 +390,6 @@ extension MobileShellComposite { + "attempt=\(attempt.id.uuidString)" ) } - if self.connectionState == .connected { - self.connectionState = .disconnected - self.macConnectionStatus = .unavailable - self.clearRemoteConnectionContext() - } self.applyConnectionRecoveryOwnerState() // Recovery uses authenticated local Iroh state first. A stuck @@ -372,6 +409,11 @@ extension MobileShellComposite { outcome: reconnectOutcome, connectionGeneration: self.connectionGeneration ) else { return } + if !reconnectOutcome.didConnect { + self.connectionState = .disconnected + self.macConnectionStatus = .unavailable + self.clearRemoteConnectionContext() + } self.applyConnectionRecoveryOwnerState() } onCancel: { MobileDebugLog.anchormux( @@ -405,15 +447,16 @@ extension MobileShellComposite { try? await ContinuousClock().sleep(for: .seconds(seconds)) guard let self, !Task.isCancelled else { return } guard self.connectionRecoveryOwner.isCurrent(attempt), - self.connectionRecoveryOwner.isRedialingOrValidating, - self.connectionState != .connected else { return } + self.connectionRecoveryOwner.isRedialingOrValidating else { return } MobileDebugLog.anchormux( "connection.recovery attempt deadline forced failure " + "trigger=\(attempt.trigger) attempt=\(attempt.id.uuidString)" ) guard self.connectionRecoveryOwner.failReplacement() != nil else { return } self.recordConnectionRecoveryFailed(attempt, failure: .timedOut) + self.connectionState = .disconnected self.macConnectionStatus = .unavailable + self.clearRemoteConnectionContext() self.applyConnectionRecoveryOwnerState() self.armAutomaticReconnectRetryAfterFailedAttempt( failure: .timedOut, @@ -555,10 +598,11 @@ extension MobileShellComposite { case .redialing, .validatingReplacement: isRecoveringConnection = true connectionRecoveryFailed = false - if connectionState == .connected { markMacConnectionReconnecting() } + markMacConnectionReconnecting() case .failed: isRecoveringConnection = false connectionRecoveryFailed = true + if connectionState != .connected { macConnectionStatus = .unavailable } } } diff --git a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift index 22d9a82e8d58..dce8f673b8b4 100644 --- a/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift +++ b/Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift @@ -1243,7 +1243,6 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { private var renderGridLivenessProbeTask: Task? private var renderGridLivenessProbeID: UUID? private var renderGridLivenessConsecutiveProbeFailures = 0 - private var renderGridLivenessLaneRepairAttempts = 0 var lastTerminalEventAt: Date? @ObservationIgnored var terminalInputAckResubscribeRetryTask: Task? @ObservationIgnored var terminalInputAckResubscribeRetryTaskID: UUID? @@ -11102,6 +11101,26 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { await replaceRemoteClientAwaitingTeardownRegistration(with: nil) } + /// Retire the dead transport synchronously while retaining the Mac identity + /// and workspace rows. Recovery owns the visible status; connectionState + /// still records real transport availability so streams restart on adoption. + func retireRemoteClientForConnectionRecovery() { + connectionGeneration = UUID() + connectionAttemptGeneration = UUID() + cancelRemoteOperationTasks() + macConnectionStatus = .reconnecting + connectionState = .disconnected + rawTerminalInputBuffer.clear() + terminalInputRPCPipeline.clear() + resumeRawTerminalInputDrainWaiters() + if let focused = focusedForegroundConnection, + focused.client === remoteClient { + removeControlCapability(ifMatching: focused) + removeFocusedConnection(ifMatching: focused) + } + replaceRemoteClient(with: nil) + } + /// Retire the current pre-authentication candidate before a newer connect /// competes for the same physical route. The registry lease transfers to /// teardown before the replacement reaches transport admission. @@ -11996,7 +12015,10 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { } func markMacConnectionReconnecting() { - guard connectionState == .connected, remoteClient != nil else { + // An active replacement owns the presentation while its transport is + // absent. Probes on a usable connection never enter this phase. + guard connectionState == .connected + || connectionRecoveryOwner.isRedialingOrValidating else { macConnectionStatus = .unavailable return } @@ -12029,8 +12051,8 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { } /// Applies an availability failure only to the connection that produced it. - /// A blocked transport write is definitive and enters the single recovery - /// owner; an ordinary response timeout remains scoped to that one RPC. + /// Request deadlines remain scoped to their operation. A transport failure + /// enters recovery only after checking the native connection state. func handleMacAvailabilityFailureIfCurrent( after error: any Error, expectedClient: MobileCoreRPCClient, @@ -13993,7 +14015,7 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { mobileShellLog.info("terminal event stream ended before subscribe ack, marking unavailable") MobileDebugLog.anchormux("sync.stream_ended before subscribe ack; failed start") diagnosticLog?.record(DiagnosticEvent(.error)) - recoverDeadConnection( + recoverClosedControlSession( trigger: .subscriptionStartFailed, expectedClient: client ) @@ -14066,7 +14088,6 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { renderGridLivenessProbeTask = nil renderGridLivenessProbeID = nil renderGridLivenessConsecutiveProbeFailures = 0 - renderGridLivenessLaneRepairAttempts = 0 } /// Single ownership point for the liveness clock the watchdog reads. @@ -14081,7 +14102,6 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { private func recordTerminalEventStreamLiveness() { lastTerminalEventAt = runtime?.now() ?? Date() renderGridLivenessConsecutiveProbeFailures = 0 - renderGridLivenessLaneRepairAttempts = 0 } #if DEBUG @@ -14208,19 +14228,13 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { // stream stalled. Keep the shared Iroh session, whose other // lanes may still carry terminal input and keepalives, and // restart only the event listener. - self.renderGridLivenessLaneRepairAttempts += 1 - let laneRepairAttempts = self.renderGridLivenessLaneRepairAttempts - let escalate = laneRepairAttempts >= 2 - if escalate { - self.renderGridLivenessLaneRepairAttempts = 0 - } MobileDebugLog.anchormux( - "sync.liveness event_lane_repair transport_alive attempts=\(laneRepairAttempts) escalate=\(escalate) silentMs=\(silentMs)" + "sync.liveness event_lane_repair transport_alive silentMs=\(silentMs)" ) self.resyncTerminalOutput( - reason: escalate ? "liveness_event_lane_escalated" : "liveness_event_lane", + reason: "liveness_event_lane", restartEventStream: true, - recoversConnectionOnSubscriptionFailure: escalate + recoversConnectionOnSubscriptionFailure: false ) return } @@ -14232,9 +14246,7 @@ public final class MobileShellComposite: MobileTerminalOutputSinking { ) self.diagnosticLog?.record(DiagnosticEvent(.livenessResubscribe, ms: UInt32(clamping: silentMs))) mobileShellLog.info("render-grid stream silent for \(silentMs, privacy: .public)ms and subscription probe failed, re-subscribing") - // The bounded probe proved this exact client dead. Hand the session - // to the single recovery owner instead of rebuilding another listener - // on the same stale shell. + // Confirm native closure before handing the session to recovery. self.recoverDeadConnection(trigger: .liveness, expectedClient: client) } } diff --git a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/IrohConnectionRecoveryOwnerTests.swift b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/IrohConnectionRecoveryOwnerTests.swift index 9e8e6df79c00..e39a09c5b35b 100644 --- a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/IrohConnectionRecoveryOwnerTests.swift +++ b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/IrohConnectionRecoveryOwnerTests.swift @@ -12,7 +12,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let initialReconnectGeneration = fixture.store.storedMacReconnectGeneration let firstClient = try #require(fixture.store.remoteClient) let first = try #require(fixture.box.get()) @@ -40,7 +40,7 @@ extension ReconnectRouteSelectionTests { await fixture.router.setWorkspaceIDs(["cmux-master", "hevy-cli"]) #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let firstClient = try #require(fixture.store.remoteClient) let hevyWorkspace = try #require( fixture.store.workspaces.first { $0.rpcWorkspaceID.rawValue == "hevy-cli" } @@ -69,7 +69,7 @@ extension ReconnectRouteSelectionTests { } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let firstClient = try #require(fixture.store.remoteClient) fixture.store.recoverDeadConnection( @@ -78,6 +78,12 @@ extension ReconnectRouteSelectionTests { ) #expect(await closeGate.waitUntilCloseStarted()) + #expect(fixture.store.connectionState == .disconnected) + #expect(fixture.store.remoteClient == nil) + #expect(fixture.store.foregroundMacDeviceID == "test-mac") + #expect(!fixture.store.workspaces.isEmpty) + #expect(fixture.store.macConnectionStatus == .reconnecting) + #expect(fixture.store.isRecoveringConnection) #expect(fixture.factory.attemptedKinds() == [.iroh]) await closeGate.release() @@ -95,7 +101,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let firstClient = try #require(fixture.store.remoteClient) fixture.store.suspendForegroundRefresh() fixture.clock.advance(by: 61) @@ -237,7 +243,7 @@ extension ReconnectRouteSelectionTests { } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let firstClient = try #require(fixture.store.remoteClient) let backing = try #require(fixture.store.pairedMacStore as? BackingUpPairedMacStore) await backup.blockFutureFetches() @@ -263,7 +269,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let first = try #require(fixture.box.get()) await first.close() @@ -282,7 +288,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let firstClient = try #require(fixture.store.remoteClient) // Keep every post-drop dial failing until the first owner reaches its // terminal failed phase. This avoids coupling the assertion to a @@ -329,7 +335,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let client = try #require(fixture.store.remoteClient) let generation = fixture.store.connectionGeneration @@ -350,7 +356,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let liveClient = try #require(fixture.store.remoteClient) let liveGeneration = fixture.store.connectionGeneration @@ -368,7 +374,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let foregroundKey = fixture.store.foregroundMacKey var staleState = try #require(fixture.store.workspacesByMac[foregroundKey]) staleState.status = .unavailable @@ -398,32 +404,47 @@ extension ReconnectRouteSelectionTests { ) } - @Test func stalledWriteRedialsExactCurrentIrohClientOnce() async throws { + @Test func failedNetworkRefreshPreservesOpenIrohConnection() async throws { let fixture = try await makeRecoveryOwnerFixture() defer { fixture.release() } - #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) let client = try #require(fixture.store.remoteClient) let generation = fixture.store.connectionGeneration + fixture.store.stateSyncAuthorityClientID = nil + let nextList = await fixture.router.count(of: "mobile.workspace.list") + 1 + await fixture.router.failWorkspaceListRequest(number: nextList) + fixture.store.recoverMobileConnection(trigger: .networkChange) + #expect(try await pollUntil { !fixture.store.connectionRecoveryOwner.isActive }) + #expect(fixture.store.remoteClient === client) + #expect(fixture.store.connectionGeneration == generation) + #expect(fixture.store.macConnectionStatus == .connected) + #expect(!fixture.store.isRecoveringConnection) + #expect(await fixture.store.reloadWorkspaceListFromMac()) + #expect(fixture.factory.attemptedKinds() == [.iroh]) + } + @Test func operationFailurePreservesOpenIrohConnection() async throws { + let fixture = try await makeRecoveryOwnerFixture() + defer { fixture.release() } + #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) + #expect(try await pollUntil { !fixture.store.isRecoveringConnection }) + let client = try #require(fixture.store.remoteClient) + let generation = fixture.store.connectionGeneration fixture.store.handleMacAvailabilityFailureIfCurrent( after: MobileShellConnectionError.transportWriteTimedOut, expectedClient: client, expectedGeneration: generation ) - - #expect(try await pollUntil { - guard let replacement = fixture.store.remoteClient else { return false } - let subscribeCount = await fixture.router.count(of: "mobile.events.subscribe") - return replacement !== client - && fixture.store.connectionState == .connected - && fixture.store.activeRoute?.kind == .iroh - && fixture.store.macConnectionStatus == .connected - && fixture.store.isRecoveringConnection == false - && subscribeCount >= 2 - }) - #expect(fixture.factory.attemptedKinds() == [.iroh, .iroh]) + // Exercise the connection after the failure, so preservation also + // proves that the installed client continues serving real requests. + #expect(await fixture.store.reloadWorkspaceListFromMac()) + #expect(fixture.store.remoteClient === client) + #expect(fixture.store.connectionGeneration == generation) + #expect(fixture.store.connectionState == .connected) + #expect(!fixture.store.isRecoveringConnection) + #expect(fixture.factory.attemptedKinds() == [.iroh]) } @Test func failedIrohRecoveryPreservesTheTypedFailureCategory() async throws { @@ -431,7 +452,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) await fixture.diagnosticLog.clear() fixture.factory.setConnectFailure(.timedOut) let first = try #require(fixture.box.get()) @@ -492,7 +513,7 @@ extension ReconnectRouteSelectionTests { defer { fixture.release() } #expect(await fixture.store.reconnectActiveMacIfAvailable(stackUserID: "user-1")) - #expect(await fixture.router.waitForCount(of: "mobile.events.subscribe", atLeast: 1)) + #expect(try await pollUntil { fixture.store.lastSuccessfulTerminalSubscription != nil }) await fixture.diagnosticLog.clear() let client = try #require(fixture.store.remoteClient) let generation = fixture.store.connectionGeneration diff --git a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellForegroundResumeTests.swift b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellForegroundResumeTests.swift index 30bf050c8297..528442a06e5f 100644 --- a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellForegroundResumeTests.swift +++ b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellForegroundResumeTests.swift @@ -133,7 +133,7 @@ struct MobileShellForegroundConnectionRecoveryTests { } @MainActor -@Test func failedForegroundProbeStillSurfacesRedialRecoveryState() async throws { +@Test func failedForegroundProbeWithLiveTransportDoesNotRedial() async throws { let router = LivenessHostRouter() let box = TransportBox() let clock = TestClock() @@ -152,14 +152,55 @@ struct MobileShellForegroundConnectionRecoveryTests { store.resumeForegroundRefresh() #expect(try await pollUntil { - if case .redialing = store.connectionRecoveryOwner.phase { - return store.isRecoveringConnection - } - return false + store.connectionRecoveryOwner.phase == .idle + && !store.isRecoveringConnection + && store.connectionState == .connected }) await router.releaseAllHeld() } +@MainActor +@Test func foregroundProbeTimeoutWithLiveTransportRepairsMountedTerminal() async throws { + let router = LivenessHostRouter() + let box = TransportBox() + let clock = TestClock() + let (store, directory) = try await makeForegroundRecoveryStore( + router: router, + box: box, + clock: clock, + probeTimeoutNanoseconds: 50_000_000 + ) + defer { + Task { await router.releaseAllHeld() } + try? FileManager.default.removeItem(at: directory) + } + let collector = OutputCollector() + collector.mount(store: store, surfaceID: "live-terminal") + await router.waitForCount(of: "mobile.terminal.replay", atLeast: 1) + try await waitForReplayResponsesServed( + 1, + router: router, + "the cold replay response must settle before testing a foreground probe timeout" + ) + let replayCount = await router.count(of: "mobile.terminal.replay") + let originalTransport = try #require(box.get()) + + store.suspendForegroundRefresh() + clock.advance(by: 31) + await router.holdNextWorkspaceListRequests() + store.resumeForegroundRefresh() + #expect(await router.waitForCount(of: "mobile.sync.fetch", atLeast: 2)) + + #expect(await router.waitForCount( + of: "mobile.terminal.replay", + atLeast: replayCount + 1, + timeoutNanoseconds: 1_000_000_000 + )) + #expect(store.connectionState == .connected) + #expect(box.get() === originalTransport) + collector.unmount() +} + @MainActor @Test func suspendingForegroundRefreshCancelsInFlightProbeWithoutRedial() async throws { let router = LivenessHostRouter() diff --git a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellRenderGridLivenessTests.swift b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellRenderGridLivenessTests.swift index c1eaf28b97cc..a8093f3fec87 100644 --- a/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellRenderGridLivenessTests.swift +++ b/Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellRenderGridLivenessTests.swift @@ -576,12 +576,12 @@ import Testing clock.advance(by: 10) store.debugRunRenderGridLivenessCheckForTesting() #expect(await router.waitForCount(of: "mobile.events.probe", atLeast: 1)) - // The second independent probe succeeds and resets the liveness window. - // Reconnecting makes `.connected` prove that its response was applied. + // The idempotent subscription retry succeeds on the same connection. + // The status change proves that its acknowledgement was applied. store.markMacConnectionReconnecting() let followUpSucceeded = try await pollUntil { store.debugRunRenderGridLivenessCheckForTesting() - return await router.count(of: "mobile.events.probe") >= 2 + return await router.count(of: "mobile.events.subscribe") >= 2 && store.macConnectionStatus == .connected } #expect(followUpSucceeded, "the follow-up probe must complete successfully") @@ -623,7 +623,7 @@ import Testing store.debugRunRenderGridLivenessCheckForTesting() let probeCount = await router.count(of: "mobile.events.probe") let subscribeCount = await router.count(of: "mobile.events.subscribe") - return probeCount >= 2 && subscribeCount >= 2 + return probeCount >= 1 && subscribeCount >= 2 } #expect(repaired, "a live transport must restart only the stalled event listener") #expect(store.remoteClient === originalClient) diff --git a/Sources/Mobile/MobileHostConnectionEventQueue.swift b/Sources/Mobile/MobileHostConnectionEventQueue.swift index c6422fb3655e..6577ed140ba4 100644 --- a/Sources/Mobile/MobileHostConnectionEventQueue.swift +++ b/Sources/Mobile/MobileHostConnectionEventQueue.swift @@ -21,10 +21,8 @@ import Foundation /// - `terminal.updated` / `workspace.updated`: level-triggered pings; the newer /// occurrence that forced the shed supersedes the shed one. /// -/// Every other topic keeps the close-on-overflow contract: those payloads -/// cannot be re-derived by the client, so tearing the connection down (the -/// client reconnects and re-syncs from authoritative state) is the only -/// lossless bound. +/// Other topics retain their ordered payloads even beyond the shedding budget. +/// Congestion is not evidence that the connection has closed. enum MobileHostEventTopicPolicy { static let renderGridTopic = "terminal.render_grid" static let simulatorFrameTopic = "simulator.frame" @@ -33,7 +31,7 @@ enum MobileHostEventTopicPolicy { switch topic { case renderGridTopic: // A render-grid event without a surface key cannot be resynced - // per-surface, so it keeps the lossless close-on-overflow path. + // per-surface, so its ordered payload is retained. return coalesceKey != nil case simulatorFrameTopic: // Simulator frames are whole-screen snapshots; a later frame fully @@ -49,13 +47,10 @@ enum MobileHostEventTopicPolicy { /// Outcome of one synchronous admission attempt on a connection's event queue. struct MobileHostEventEnqueueResult: Sendable { - /// The event was appended to the bounded queue. + /// The event was appended to the queue. let admitted: Bool /// The caller must start the (single) drain task for this connection. let startDrain: Bool - /// A non-droppable event overflowed the bounded queue; the caller must - /// close the connection — the lossless-topic contract. - let shouldClose: Bool /// Surfaces whose queued render-grid frames were shed; the caller must ask /// the producer for a full-frame resync of each. let renderGridResyncSurfaceIDs: Set @@ -71,7 +66,6 @@ struct MobileHostEventEnqueueResult: Sendable { static let rejected = MobileHostEventEnqueueResult( admitted: false, startDrain: false, - shouldClose: false, renderGridResyncSurfaceIDs: [], depthAfterEnqueue: nil, shedEventCount: 0, @@ -95,18 +89,10 @@ private struct MobileHostEventShedSummary: Sendable { } } -/// Bounded, synchronously-admitted mailbox between the event fan-out -/// (``MobileHostService/emitEvent(topic:payload:)``) and one connection's -/// drain loop. -/// -/// This is the owner boundary for issue #8842: admission runs *before* any -/// per-connection work is scheduled, on the emitter's thread, against an -/// explicit bound — so the memory pinned per connection is O(capacity) no -/// matter how far emission runs ahead of a slow, paused, or half-dead -/// subscriber. The previous path spawned one unstructured Task per connection -/// per event, each retaining the full payload dictionary until the connection -/// actor reached its own bounded check, which left everything upstream of the -/// bound unbounded. +/// Synchronously admitted mailbox between event fan-out and a single drain. +/// Refresh events have a shedding budget; ordered events are retained until +/// delivery. Admission happens before task creation, so producers never create +/// a separate task retaining each event while the network is slow. final class MobileHostConnectionEventQueue: @unchecked Sendable { struct QueuedEvent: Sendable { let topic: String @@ -174,7 +160,7 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { return subscribedTopics.contains(topic) } - /// Synchronous bounded admission. Safe to call from any thread; never + /// Synchronous admission with refresh-event shedding. Safe on any thread; never /// blocks on the network, the connection actor, or the runtime. func enqueue( topic: String, @@ -214,7 +200,6 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { return MobileHostEventEnqueueResult( admitted: false, startDrain: false, - shouldClose: false, renderGridResyncSurfaceIDs: resyncSurfaceIDs, depthAfterEnqueue: nil, shedEventCount: shedSummary.eventCount, @@ -222,20 +207,8 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { simulatorFrameShedPanelIDs: shedSummary.simulatorFramePanelIDs ) } - guard hasRoomLocked(for: frame) else { - guard MobileHostEventTopicPolicy.isDroppable(topic: topic, coalesceKey: coalesceKey) else { - lock.unlock() - return MobileHostEventEnqueueResult( - admitted: false, - startDrain: false, - shouldClose: true, - renderGridResyncSurfaceIDs: resyncSurfaceIDs, - depthAfterEnqueue: nil, - shedEventCount: shedSummary.eventCount, - shedByteCount: shedSummary.byteCount, - simulatorFrameShedPanelIDs: shedSummary.simulatorFramePanelIDs - ) - } + if !hasRoomLocked(for: frame), + MobileHostEventTopicPolicy.isDroppable(topic: topic, coalesceKey: coalesceKey) { if isRenderGrid, let coalesceKey { if poisonedRenderGridSurfaceIDs.insert(coalesceKey).inserted { resyncSurfaceIDs.insert(coalesceKey) @@ -249,7 +222,6 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { return MobileHostEventEnqueueResult( admitted: false, startDrain: false, - shouldClose: false, renderGridResyncSurfaceIDs: resyncSurfaceIDs, depthAfterEnqueue: nil, shedEventCount: shedSummary.eventCount, @@ -279,7 +251,6 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { return MobileHostEventEnqueueResult( admitted: true, startDrain: startDrain, - shouldClose: false, renderGridResyncSurfaceIDs: resyncSurfaceIDs, depthAfterEnqueue: depthAfterEnqueue, shedEventCount: shedSummary.eventCount, diff --git a/Sources/Mobile/MobileHostIrxRuntime.swift b/Sources/Mobile/MobileHostIrxRuntime.swift index 7e0b856f717c..0b6d6e0c3060 100644 --- a/Sources/Mobile/MobileHostIrxRuntime.swift +++ b/Sources/Mobile/MobileHostIrxRuntime.swift @@ -595,9 +595,8 @@ final class MobileHostIrxRuntime: MobileHostPairingRuntime { independentEventWriter: eventWriter, // Admission has already authenticated this bounded pooled peer. // It may wait for its first RPC while the client finishes setup; - // transport keepalives and the pool own its connection lifetime. + // native Iroh owns its connection lifetime. firstFrameTimeoutNanoseconds: 0, - idleTimeoutNanoseconds: 0, isCurrent: { [weak self] in let runtime = self return await MainActor.run { runtime?.isCurrent(token) == true } diff --git a/Sources/Mobile/MobileHostService.swift b/Sources/Mobile/MobileHostService.swift index 8fec55383ecd..fc4d60cbca17 100644 --- a/Sources/Mobile/MobileHostService.swift +++ b/Sources/Mobile/MobileHostService.swift @@ -662,17 +662,7 @@ final class MobileHostService { if result.startDrain { Task { await connection.drainQueuedEvents() } } - if result.shouldClose { - Task { - await connection.close( - reason: "event queue exceeded bounded capacity", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlWriteFailed, - failure: .sendQueueOverflow - ) - ) - } - } + } if !resyncSurfaceIDs.isEmpty { MobileTerminalRenderObserver.requestRenderGridFullResync( @@ -854,7 +844,6 @@ final class MobileHostService { artifactTransfers: MobileHostIrohArtifactTransferRegistry? = nil, independentEventWriter: (any MobileHostIndependentEventWriting)? = nil, firstFrameTimeoutNanoseconds: UInt64? = nil, - idleTimeoutNanoseconds: UInt64? = nil, promoteUsableSession: @escaping @Sendable () async -> Bool = { true }, remoteControlDisabledByPolicy: @escaping @Sendable () -> Bool = { MobileRemoteControlPolicy.isDisabled @@ -881,13 +870,20 @@ final class MobileHostService { } let id = UUID() + let defaultFirstFrameTimeout: UInt64 = switch authorization { + case .irohAdmission: + // Iroh owns admission and native connection liveness. A delayed + // first control frame is valid while the admitted session is + // settling, so an application timer must not retire it. + 0 + case .stackBearer: + MobileHostConnection.defaultFirstFrameTimeoutNanoseconds + } let session = MobileHostConnection( id: id, transport: transport, firstFrameTimeoutNanoseconds: firstFrameTimeoutNanoseconds - ?? MobileHostConnection.defaultFirstFrameTimeoutNanoseconds, - idleTimeoutNanoseconds: idleTimeoutNanoseconds - ?? MobileHostConnection.defaultIdleTimeoutNanoseconds, + ?? defaultFirstFrameTimeout, independentEventWriter: independentEventWriter, authorizeRequest: { request in await Self.connectionAuthorizationError( @@ -1346,15 +1342,7 @@ extension MobileHostService { #endif actor MobileHostConnection { - private static let maximumReceiveBufferByteCount = MobileSyncFrameCodec.defaultMaximumFrameByteCount + MobileSyncFrameCodec.headerByteCount fileprivate static let defaultFirstFrameTimeoutNanoseconds: UInt64 = 15 * 1_000_000_000 - fileprivate static let defaultIdleTimeoutNanoseconds: UInt64 = 30 * 1_000_000_000 - /// Bounded deadline for one control-lane event write. A peer that accepted - /// the connection but stopped reading (TCP zero-window, QUIC flow-control - /// stall) would otherwise pin the drain — and with it this connection's - /// queue, transport, and tasks — indefinitely (issue #8842). - private static let defaultEventSendStallTimeoutNanoseconds: UInt64 = 30 * 1_000_000_000 - private struct EventSubscription: Sendable { let topics: Set let transport: MobileHostEventTransport @@ -1394,7 +1382,6 @@ actor MobileHostConnection { private let writer: MobileHostSerializedTransportWriter private let independentEventWriter: (any MobileHostIndependentEventWriting)? private let firstFrameTimeoutNanoseconds: UInt64 - private let idleTimeoutNanoseconds: UInt64 private let authorizeRequest: @Sendable (MobileHostRPCRequest) async -> MobileHostRPCResult? private let onAuthorizedRequest: @Sendable (MobileHostRPCRequest) async -> Void private let onUsableSession: @Sendable () async -> Bool @@ -1402,18 +1389,12 @@ actor MobileHostConnection { private let onClose: @Sendable (UUID) async -> Void private let requestSimulatorFrameReplay: @Sendable (UUID, Set) async -> Void private let responseWorkQuota = MobileHostRPCWorkQuota() - /// Bounded pre-write mailbox with synchronous admission from the event + /// Pre-write mailbox with synchronous admission from the event /// fan-out. Nonisolated so ``MobileHostService/emitEvent(topic:payload:)`` /// admits events without scheduling any per-event actor work. nonisolated let eventQueue: MobileHostConnectionEventQueue - private let eventSendStallTimeoutNanoseconds: UInt64 - /// Invalidates the pending event-send stall deadline: bumped when a send - /// starts and again when it settles, so a deadline armed for send N can - /// never close the connection after N completed. - private var eventSendGeneration: UInt64 = 0 private var receiveBuffer = Data() private var firstFrameTimeoutTask: Task? - private var idleTimeoutTask: Task? private var responseTasks: [UUID: ResponseTask] = [:] /// PTY-writing requests are ordered PER SURFACE: ordering is only a /// property of one terminal, and a connection-wide FIFO would let one @@ -1442,8 +1423,6 @@ actor MobileHostConnection { connection: NWConnection, eventQueue: MobileHostConnectionEventQueue = MobileHostConnectionEventQueue(), firstFrameTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultFirstFrameTimeoutNanoseconds, - idleTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultIdleTimeoutNanoseconds, - eventSendStallTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultEventSendStallTimeoutNanoseconds, independentEventWriter: (any MobileHostIndependentEventWriting)? = nil, authorizeRequest: @escaping @Sendable (MobileHostRPCRequest) async -> MobileHostRPCResult?, onAuthorizedRequest: @escaping @Sendable (MobileHostRPCRequest) async -> Void, @@ -1458,8 +1437,6 @@ actor MobileHostConnection { self.writer = MobileHostSerializedTransportWriter(transport: transport) self.independentEventWriter = independentEventWriter self.firstFrameTimeoutNanoseconds = firstFrameTimeoutNanoseconds - self.idleTimeoutNanoseconds = idleTimeoutNanoseconds - self.eventSendStallTimeoutNanoseconds = eventSendStallTimeoutNanoseconds self.authorizeRequest = authorizeRequest self.onAuthorizedRequest = onAuthorizedRequest self.onUsableSession = onUsableSession @@ -1474,8 +1451,6 @@ actor MobileHostConnection { transport: any CmxByteTransport, eventQueue: MobileHostConnectionEventQueue = MobileHostConnectionEventQueue(), firstFrameTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultFirstFrameTimeoutNanoseconds, - idleTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultIdleTimeoutNanoseconds, - eventSendStallTimeoutNanoseconds: UInt64 = MobileHostConnection.defaultEventSendStallTimeoutNanoseconds, independentEventWriter: (any MobileHostIndependentEventWriting)? = nil, authorizeRequest: @escaping @Sendable (MobileHostRPCRequest) async -> MobileHostRPCResult?, onAuthorizedRequest: @escaping @Sendable (MobileHostRPCRequest) async -> Void, @@ -1489,8 +1464,6 @@ actor MobileHostConnection { self.writer = MobileHostSerializedTransportWriter(transport: transport) self.independentEventWriter = independentEventWriter self.firstFrameTimeoutNanoseconds = firstFrameTimeoutNanoseconds - self.idleTimeoutNanoseconds = idleTimeoutNanoseconds - self.eventSendStallTimeoutNanoseconds = eventSendStallTimeoutNanoseconds self.authorizeRequest = authorizeRequest self.onAuthorizedRequest = onAuthorizedRequest self.onUsableSession = onUsableSession @@ -1567,8 +1540,6 @@ actor MobileHostConnection { self.exit = exit firstFrameTimeoutTask?.cancel() firstFrameTimeoutTask = nil - idleTimeoutTask?.cancel() - idleTimeoutTask = nil receiveTask?.cancel() receiveTask = nil // Rejects all future admissions and releases every queued payload; the @@ -1602,56 +1573,40 @@ actor MobileHostConnection { private func handleReceive(data: Data) async { if !data.isEmpty { - idleTimeoutTask?.cancel() - idleTimeoutTask = nil - guard receiveBuffer.count + data.count <= Self.maximumReceiveBufferByteCount else { - _ = await sendResponse( - MobileHostRPCEnvelope.error( - id: nil, - code: "frame_decode_error", - message: "Invalid frame" - ) - ) - await close( - reason: "receive buffer exceeded frame limit", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlReadFailed, - failure: .protocolViolation - ) - ) - return - } + // Message limits belong to individual frames. A receive chunk may + // contain the tail of a maximum-size frame followed by another. receiveBuffer.append(data) do { - let frames = try MobileSyncFrameCodec.decodeFrames( - from: &receiveBuffer, - maximumDecodedFrameCount: responseWorkQuota - .maximumConcurrentRequestCount - ) - if !frames.isEmpty { - didDecodeFirstFrame = true - firstFrameTimeoutTask?.cancel() - firstFrameTimeoutTask = nil - } - for frame in frames { - guard !isClosed else { - return + let batchLimit = responseWorkQuota.maximumConcurrentRequestCount + while !isClosed, !Task.isCancelled { + let frames = try MobileSyncFrameCodec.decodeFrames( + from: &receiveBuffer, + maximumDecodedFrameCount: batchLimit + ) + if !frames.isEmpty { + didDecodeFirstFrame = true + firstFrameTimeoutTask?.cancel() + firstFrameTimeoutTask = nil } - guard startResponseTask(for: frame) else { - await close( - reason: "rpc work capacity exceeded", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlReadFailed, - failure: .protocolViolation - ) - ) - return + for frame in frames { + guard !isClosed else { return } + if !startResponseTask(for: frame) { + // Work pressure fails this request explicitly; it + // does not invalidate the authenticated connection. + let request = try? MobileHostRPCEnvelope.decodeRequest(frame).get() + guard await sendResponse(MobileHostRPCEnvelope.error( + id: request?.id, + code: "server_busy", + message: "Too many requests are pending" + )) else { return } + } } + guard frames.count == batchLimit else { break } + await Task.yield() } guard !isClosed else { return } - startIdleTimeout() } catch { _ = await sendResponse( MobileHostRPCEnvelope.error( @@ -1747,9 +1702,6 @@ actor MobileHostConnection { startOrderedRequestWorkerIfNeeded(surfaceKey: surfaceKey) } else { orderedRequestQueuesBySurfaceKey[surfaceKey] = nil - if !hasActiveResponseWork { - startIdleTimeout() - } } } @@ -1771,18 +1723,8 @@ actor MobileHostConnection { ) } - private var hasActiveResponseWork: Bool { - !responseTasks.isEmpty - || !orderedRequestWorkerTasksBySurfaceKey.isEmpty - || !orderedRequestRunningFrameByteCountsBySurfaceKey.isEmpty - || orderedRequestQueuesBySurfaceKey.values.contains { !$0.isEmpty } - } - private func finishResponseTask(_ taskID: UUID) { responseTasks[taskID] = nil - if !hasActiveResponseWork { - startIdleTimeout() - } } private func startFirstFrameTimeout() { @@ -1812,37 +1754,6 @@ actor MobileHostConnection { ) } - private func startIdleTimeout() { - guard idleTimeoutNanoseconds > 0, - didDecodeFirstFrame, - !isClosed, - subscriptions.isEmpty, - !hasActiveResponseWork else { - return - } - idleTimeoutTask?.cancel() - let timeoutNanoseconds = idleTimeoutNanoseconds - idleTimeoutTask = Task { [weak self] in - do { - try await Task.sleep(nanoseconds: timeoutNanoseconds) - await self?.closeIfIdleAfterFrame() - } catch {} - } - } - - private func closeIfIdleAfterFrame() async { - guard didDecodeFirstFrame, subscriptions.isEmpty, !hasActiveResponseWork else { - return - } - await close( - reason: "idle after frame timed out", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlReadFailed, - failure: .timedOut - ) - ) - } - private func respond( to decodedRequest: Result ) async { @@ -2151,8 +2062,6 @@ actor MobileHostConnection { previousTopics: previousTopics, nextTopics: topics ) - idleTimeoutTask?.cancel() - idleTimeoutTask = nil if currentSubscribedTopics().contains(MobileHostEventTopicPolicy.simulatorFrameTopic) { await dispatchPendingSimulatorFrameReplay() } @@ -2178,9 +2087,6 @@ actor MobileHostConnection { }) { await resetIndependentEventWriter() } - if subscriptions.isEmpty { - startIdleTimeout() - } return removed } @@ -2241,26 +2147,12 @@ actor MobileHostConnection { if result.startDrain { Task { await self.drainQueuedEvents() } } - if result.shouldClose { - // The bounded queue fills when the control stream stops draining - // (e.g. the peer's network path died mid-write) while terminal - // events keep arriving. The peer violated nothing; field host - // rings (2026-07-23 WiFi path flap) showed this close mislabeled - // protocolViolation seconds after admission. - await close( - reason: "event queue exceeded bounded capacity", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlWriteFailed, - failure: .sendQueueOverflow - ) - ) - } return result.admitted } /// Synchronous bounded admission from the fan-out path. Never blocks and /// never schedules per-event work; the caller acts on the returned - /// outcome (drain start, overflow close, render-grid resync). + /// outcome (drain start, refresh shedding, render-grid resync). nonisolated func enqueueEventFrame( _ frame: Data, topic: String, @@ -2318,7 +2210,7 @@ actor MobileHostConnection { /// (enforced by the queue's drain claim), pulling from the bounded queue /// and writing to the negotiated lane. Exits when the queue is empty, the /// connection closes, lane negotiation pauses delivery, or a delivery - /// fails or stalls (which closes the connection). + /// fails (which closes the unusable control session). func drainQueuedEvents() async { while true { if isClosed || independentEventNegotiationInProgress { @@ -2397,38 +2289,11 @@ actor MobileHostConnection { return await sendEventControlFrame(event.frame) } - /// Writes one event frame on the control lane under the bounded stall - /// deadline. On a stall the connection is closed — `transport.close()` - /// resolves the pending write — converting a half-dead subscriber into - /// deterministic teardown instead of a forever-pinned drain. + /// Writes one serialized frame until the transport completes or fails. + /// An application deadline cannot cancel writeAll safely: it may already + /// have sent a prefix. Native transport failure still ends the drain. private func sendEventControlFrame(_ frame: Data) async -> Bool { - guard !isClosed else { return false } - let timeoutNanoseconds = eventSendStallTimeoutNanoseconds - guard timeoutNanoseconds > 0 else { - return await sendControlFrame(frame) - } - eventSendGeneration &+= 1 - let generation = eventSendGeneration - let deadlineTask = Task { [weak self] in - try? await Task.sleep(nanoseconds: timeoutNanoseconds) - guard !Task.isCancelled else { return } - await self?.closeIfEventSendStillInFlight(generation: generation) - } - let delivered = await sendControlFrame(frame) - eventSendGeneration &+= 1 - deadlineTask.cancel() - return delivered && !isClosed - } - - private func closeIfEventSendStillInFlight(generation: UInt64) async { - guard eventSendGeneration == generation, !isClosed else { return } - await close( - reason: "event send stalled past the bounded deadline", - exit: CmxIrohAdmittedConnectionExit( - lifecycle: .controlWriteFailed, - failure: .timedOut - ) - ) + await sendControlFrame(frame) } private func downgradeIndependentSubscriptionsToControl() { @@ -2510,11 +2375,6 @@ extension MobileHostConnection { startFirstFrameTimeout() } - func debugStartIdleTimeoutAfterFrameForTesting() { - didDecodeFirstFrame = true - startIdleTimeout() - } - func debugHandleReceiveDataForTesting(_ data: Data) async { await handleReceive(data: data) } diff --git a/cmuxTests/MobileHostConnectionEventLaneTests.swift b/cmuxTests/MobileHostConnectionEventLaneTests.swift index 39b220fc6149..1684310f2f54 100644 --- a/cmuxTests/MobileHostConnectionEventLaneTests.swift +++ b/cmuxTests/MobileHostConnectionEventLaneTests.swift @@ -16,14 +16,10 @@ extension MobileHostAuthorizationTests { @Test func testMobileHostConnectionClosesWhenFirstFrameTimesOut() async throws { let connectionID = UUID() let recorder = MobileHostConnectionCloseRecorder() - let connection = NWConnection( - host: NWEndpoint.Host("127.0.0.1"), - port: NWEndpoint.Port(rawValue: 9)!, - using: .tcp - ) + let transport = RecordingMobileHostByteTransport() let session = MobileHostConnection( id: connectionID, - connection: connection, + transport: transport, firstFrameTimeoutNanoseconds: 1_000_000, authorizeRequest: { _ in nil }, onAuthorizedRequest: { _ in }, @@ -43,18 +39,13 @@ extension MobileHostAuthorizationTests { let finalRecordedIDs = await recorder.recordedIDs() #expect(finalRecordedIDs == [connectionID]) } - @Test func testMobileHostConnectionClosesWhenIdleAfterFirstFrame() async throws { + @Test func testMobileHostConnectionStaysOpenWhenIdleAfterFirstFrame() async throws { let connectionID = UUID() let recorder = MobileHostConnectionCloseRecorder() - let connection = NWConnection( - host: NWEndpoint.Host("127.0.0.1"), - port: NWEndpoint.Port(rawValue: 9)!, - using: .tcp - ) + let transport = RecordingMobileHostByteTransport() let session = MobileHostConnection( id: connectionID, - connection: connection, - idleTimeoutNanoseconds: 1_000_000, + transport: transport, authorizeRequest: { _ in nil }, onAuthorizedRequest: { _ in }, handleRequest: { _ in .ok([:]) }, @@ -62,18 +53,15 @@ extension MobileHostAuthorizationTests { await recorder.record(id) } ) - await session.debugStartIdleTimeoutAfterFrameForTesting() - for _ in 0..<100 { - let recordedIDs = await recorder.recordedIDs() - if !recordedIDs.isEmpty { - break - } - try await Task.sleep(nanoseconds: 1_000_000) - } - let finalRecordedIDs = await recorder.recordedIDs() - #expect(finalRecordedIDs == [connectionID]) + let frame = try MobileSyncFrameCodec.encodeFrame( + Data(#"{"id":"status","method":"mobile.host.status","params":{}}"#.utf8) + ) + await session.debugHandleReceiveDataForTesting(frame) + try await Task.sleep(nanoseconds: 25_000_000) + #expect(await recorder.recordedIDs().isEmpty) + await session.close(reason: "test cleanup") } - @Test func testMobileHostConnectionKeepsSubscribedEventStreamPastIdleTimeout() async throws { + @Test func testMobileHostConnectionKeepsSubscribedEventStreamIdle() async throws { let connectionID = UUID() let recorder = MobileHostConnectionCloseRecorder() let connection = NWConnection( @@ -84,7 +72,6 @@ extension MobileHostAuthorizationTests { let session = MobileHostConnection( id: connectionID, connection: connection, - idleTimeoutNanoseconds: 1_000_000, authorizeRequest: { _ in nil }, onAuthorizedRequest: { _ in }, handleRequest: { _ in .ok([:]) }, @@ -93,26 +80,15 @@ extension MobileHostAuthorizationTests { } ) await session.subscribe(streamID: "events", topics: ["terminal.updated"]) - await session.debugStartIdleTimeoutAfterFrameForTesting() - // An active subscription suppresses the idle-after-frame timeout: the - // arm path early-returns without scheduling any close. Awaiting an - // actor-isolated round-trip on the connection guarantees the arm call - // was fully processed and that the connection is still alive and - // subscribed, so the recorder reflects the final state with no - // wall-clock window to race. + // Subscriptions remain usable without requiring synthetic traffic. The + // host has no application-level idle deadline after admission. #expect(await session.isSubscribed(to: "terminal.updated")) let subscribedCloseIDs = await recorder.recordedIDs() #expect(subscribedCloseIDs.isEmpty) _ = await session.unsubscribe(streamID: "events") - for _ in 0..<100 { - let recordedIDs = await recorder.recordedIDs() - if !recordedIDs.isEmpty { - break - } - try await Task.sleep(nanoseconds: 1_000_000) - } - let finalRecordedIDs = await recorder.recordedIDs() - #expect(finalRecordedIDs == [connectionID]) + try await Task.sleep(nanoseconds: 25_000_000) + #expect(await recorder.recordedIDs().isEmpty) + await session.close(reason: "test cleanup") } @Test func testDeadIndependentEventLaneFallsBackCurrentAndFutureEventsToControl() async throws { @@ -183,7 +159,7 @@ extension MobileHostAuthorizationTests { await session.close(reason: "test complete") } - @Test func testIndependentEventBackpressureClosesAtBoundedQueueCapacity() async throws { + @Test func testIndependentEventBackpressurePreservesOrderedEvents() async throws { let control = RecordingMobileHostByteTransport() let independent = TestMobileHostIndependentEventWriter( behavior: .blockAfterProbe @@ -198,10 +174,7 @@ extension MobileHostAuthorizationTests { handleRequest: { _ in .ok([:]) }, onClose: { _ in } ) - // A non-droppable topic (state-sync deltas cannot be re-derived by the - // client) keeps the close-on-overflow contract; recoverable topics like - // terminal.render_grid are shed instead — see - // testStalledRenderGridSubscriberStaysOpenWithBoundedEventQueue. + // Ordered state changes remain queued while the event lane is busy. _ = await session.debugHandleSubscriptionRPCForTesting( MobileHostRPCRequest( id: "subscribe", @@ -232,15 +205,16 @@ extension MobileHostAuthorizationTests { ) } #expect( - !(await session.sendEvent( + await session.sendEvent( topic: "mobile.sync.delta", payload: ["seq": 257] - )) + ) ) - #expect(await control.observedCloseCount() == 1) - #expect(await independent.observedCloseCount() == 1) - #expect(await session.debugQueuedEventCountForTesting() == 0) + #expect(await control.observedCloseCount() == 0) + #expect(await independent.observedCloseCount() == 0) + #expect(await session.debugQueuedEventCountForTesting() == 257) + await session.close(reason: "test complete") } @Test func testIdempotentSubscriptionDoesNotReprobeHealthyIndependentLane() async throws { @@ -333,11 +307,8 @@ extension MobileHostAuthorizationTests { #expect(await transport.observedCloseCount() == 1) } - /// Events that cannot be re-derived by the client (state-sync deltas and - /// other non-refresh topics) must keep the close-on-overflow contract: the - /// host may never silently drop them, so a subscriber that stops draining - /// is torn down at the bounded capacity instead of growing without bound. - @Test func testStalledSubscriberOverflowOnNonRecoverableTopicClosesConnection() async throws { + /// Ordered events must survive congestion without forcing a reconnect. + @Test func testStalledSubscriberPreservesOrderedEventsBeyondSheddingBudget() async throws { let transport = StalledSendMobileHostByteTransport() let session = MobileHostConnection( id: UUID(), @@ -359,9 +330,10 @@ extension MobileHostAuthorizationTests { } } - #expect(await transport.observedCloseCount() == 1) - #expect(admitted <= 258) - #expect(await session.debugQueuedEventCountForTesting() == 0) + #expect(await transport.observedCloseCount() == 0) + #expect(admitted == 300) + #expect(await session.debugQueuedEventCountForTesting() >= 299) + await session.close(reason: "test complete") } /// End-to-end fan-out proof for issue #8842: sustained emission through the @@ -460,18 +432,14 @@ extension MobileHostAuthorizationTests { )) } - /// A subscriber whose transport accepted a frame but never completes the - /// write (TCP zero-window peer) is torn down by the bounded event-send - /// stall deadline instead of pinning the connection's queue, tasks, and - /// socket forever. - @Test func testEventSendStallDeadlineClosesStalledConnection() async throws { + /// A stalled event write does not end the connection or interrupt framing. + @Test func testEventSendStallDoesNotCloseConnection() async throws { let transport = StalledSendMobileHostByteTransport() let recorder = MobileHostConnectionCloseRecorder() let connectionID = UUID() let session = MobileHostConnection( id: connectionID, transport: transport, - eventSendStallTimeoutNanoseconds: 5_000_000, authorizeRequest: { _ in nil }, onAuthorizedRequest: { _ in }, handleRequest: { _ in .ok([:]) }, @@ -485,12 +453,12 @@ extension MobileHostAuthorizationTests { payload: ["surface_id": "surface-stall-8842", "full": true] ) await transport.waitUntilSendStalled() - for _ in 0..<2_000 { - if !(await recorder.recordedIDs().isEmpty) { break } - try await Task.sleep(nanoseconds: 1_000_000) - } - #expect(await recorder.recordedIDs() == [connectionID]) - #expect(await transport.observedCloseCount() == 1) + // Exercise an unresolved send across suspension before verifying + // that the connection has not been closed on the application's behalf. + try await Task.sleep(for: .milliseconds(30)) + #expect(await recorder.recordedIDs().isEmpty) + #expect(await transport.observedCloseCount() == 0) + await session.close(reason: "test complete") } // MARK: - Bounded event queue admission policy @@ -518,7 +486,6 @@ extension MobileHostAuthorizationTests { isFullRenderGridFrame: false, frame: frame ) #expect(!overflow.admitted) - #expect(!overflow.shouldClose) #expect(overflow.renderGridResyncSurfaceIDs == ["s1"]) #expect(queue.count == 0) // While poisoned, deltas stay refused even though there is room. @@ -540,7 +507,7 @@ extension MobileHostAuthorizationTests { #expect(queue.count == 2) } - @Test func testEventQueueOverflowOnNonDroppableTopicRequestsClose() { + @Test func testEventQueuePreservesOrderedEventsBeyondSheddingBudget() { let queue = MobileHostConnectionEventQueue( maximumEventCount: 1, maximumByteCount: 1_000_000 @@ -555,8 +522,10 @@ extension MobileHostAuthorizationTests { topic: "mobile.sync.delta", coalesceKey: nil, isFullRenderGridFrame: false, frame: frame ) - #expect(!overflow.admitted) - #expect(overflow.shouldClose) + #expect(overflow.admitted) + #expect(queue.count == 2) + #expect(queue.dequeue()?.frame == frame) + #expect(queue.dequeue()?.frame == frame) } @Test func testEventQueueEnforcesByteBudgetBySheddingOldestDroppable() { diff --git a/cmuxTests/MobileHostConnectionLifecycleTests.swift b/cmuxTests/MobileHostConnectionLifecycleTests.swift index 00563d7d3edb..b1879cd868c6 100644 --- a/cmuxTests/MobileHostConnectionLifecycleTests.swift +++ b/cmuxTests/MobileHostConnectionLifecycleTests.swift @@ -189,7 +189,7 @@ extension MobileHostAuthorizationTests { service.debugResetMobileLifecycleStateForTesting() } - @Test func testIrohTransportCanDisableTheControlIdleTimeout() async throws { + @Test func testMobileHostTransportStaysOpenWhenIdleAfterAdmission() async throws { let service = MobileHostService.shared service.debugResetMobileLifecycleStateForTesting() let registry = MobileHostConnectionRegistry.shared @@ -200,33 +200,12 @@ extension MobileHostAuthorizationTests { service.debugResetMobileLifecycleStateForTesting() } - let expiringTransport = ScriptedMobileHostByteTransport() let authorization = try irohAdmissionContext() - let expiringTask = Task { - await MobileHostService.acceptTransport( - expiringTransport, - authorization: authorization, - idleTimeoutNanoseconds: 1_000_000, - isCurrent: { true } - ) - } - await waitForMobileHostConnectionCount(1) - try await expiringTransport.enqueue(Self.mobileHostStatusFrame(id: "expiring")) - _ = await expiringTransport.waitForSentBufferCount(1) - await expiringTransport.waitForCloseCount(1) - #expect( - await expiringTask.value == CmxIrohAdmittedConnectionExit( - lifecycle: .controlReadFailed, - failure: .timedOut - ) - ) - let persistentTransport = ScriptedMobileHostByteTransport() let persistentTask = Task { await MobileHostService.acceptTransport( persistentTransport, authorization: authorization, - idleTimeoutNanoseconds: 0, isCurrent: { true } ) } @@ -266,7 +245,6 @@ extension MobileHostAuthorizationTests { transport, authorization: authorization, firstFrameTimeoutNanoseconds: 0, - idleTimeoutNanoseconds: 0, isCurrent: { true } ) } @@ -562,7 +540,6 @@ extension MobileHostAuthorizationTests { let session = MobileHostConnection( id: connectionID, connection: socket.connection, - idleTimeoutNanoseconds: 1_000_000, authorizeRequest: { _ in .failure(MobileHostRPCError(code: "unauthorized", message: "no")) }, @@ -577,16 +554,9 @@ extension MobileHostAuthorizationTests { ) await session.debugHandleReceiveDataForTesting(frame) try await Task.sleep(nanoseconds: 25_000_000) - await session.debugStartIdleTimeoutAfterFrameForTesting() - for _ in 0..<100 { - let recordedIDs = await recorder.recordedIDs() - if !recordedIDs.isEmpty { - break - } - try await Task.sleep(nanoseconds: 1_000_000) - } - let finalRecordedIDs = await recorder.recordedIDs() - #expect(finalRecordedIDs == [connectionID]) + try await Task.sleep(nanoseconds: 25_000_000) + #expect(await recorder.recordedIDs().isEmpty) + await session.close(reason: "test cleanup") } @Test func testMobileHostConnectionStopsBatchedFrameProcessingAfterClose() async throws { let connectionID = UUID() @@ -650,7 +620,7 @@ extension MobileHostAuthorizationTests { let recordedMethods = await requestRecorder.recordedMethods() #expect(recordedMethods == ["workspace.list"]) } - @Test func testMobileHostConnectionClosesBeforeStartingAnUnboundedRPCBatch() async throws { + @Test func testMobileHostConnectionProcessesLargeBatchWithoutDisconnecting() async throws { let transport = RecordingMobileHostByteTransport() let invocationRecorder = MobileHostAuthorizationInvocationRecorder() let session = MobileHostConnection( @@ -674,9 +644,36 @@ extension MobileHostAuthorizationTests { await session.debugHandleReceiveDataForTesting(batch) - #expect(await transport.observedCloseCount() == 1) - #expect(await invocationRecorder.count() == 0) + let responses = await transport.waitForSentBufferCount( + MobileHostRPCWorkQuota.recommendedMaximumConcurrentRequestCount + 1 + ) + #expect(responses.count == MobileHostRPCWorkQuota.recommendedMaximumConcurrentRequestCount + 1) + #expect(await transport.observedCloseCount() == 0) + await session.close(reason: "test complete") + } + @Test func testMobileHostAcceptsCoalescedTailOfMaximumSizeFrame() async throws { + let transport = RecordingMobileHostByteTransport() + let session = MobileHostConnection( + id: UUID(), transport: transport, + authorizeRequest: { _ in nil }, + onAuthorizedRequest: { _ in }, + handleRequest: { _ in .ok([:]) }, + onClose: { _ in } + ) + var payload = Data(#"{"id":"large","method":"workspace.list","params":{}}"#.utf8) + payload.append(Data(repeating: 0x20, count: MobileSyncFrameCodec.defaultMaximumFrameByteCount - payload.count)) + let frame = try MobileSyncFrameCodec.encodeFrame(payload) + await session.debugHandleReceiveDataForTesting(Data(frame.dropLast())) + var tail = Data(frame.suffix(1)) + tail.append(try MobileSyncFrameCodec.encodeFrame( + Data(#"{"id":"following","method":"workspace.list","params":{}}"#.utf8) + )) + await session.debugHandleReceiveDataForTesting(tail) + try #require(await transport.observedCloseCount() == 0) + #expect(await transport.waitForSentBufferCount(2).count == 2) + await session.close(reason: "test complete") } + // MARK: - Advertised mobile host capabilities @Test func testMobileHostAdvertisesWorkspaceActionCapabilities() { let capabilities = MobileHostService.mobileHostCapabilities diff --git a/cmuxTests/MobileHostIrohAdmissionTests.swift b/cmuxTests/MobileHostIrohAdmissionTests.swift index de5958d92205..81ead03f0eec 100644 --- a/cmuxTests/MobileHostIrohAdmissionTests.swift +++ b/cmuxTests/MobileHostIrohAdmissionTests.swift @@ -188,7 +188,6 @@ struct IrohTailscaleVersionSkewMacGateTests { id: UUID(), transport: transport, firstFrameTimeoutNanoseconds: 0, - idleTimeoutNanoseconds: 0, authorizeRequest: { request in await MobileHostService.connectionAuthorizationError( for: request, @@ -567,7 +566,13 @@ extension MobileHostAuthorizationTests { @Test func testIrohApplicationLaneQuotasReserveArtifactCapacity() { #expect(MobileHostIrohApplicationLaneRouter.maximumConcurrentTerminalLaneCount == 4) #expect(MobileHostIrohApplicationLaneRouter.maximumConcurrentArtifactLaneCount == 1) - #expect(MobileHostIrohApplicationLaneRouter.maximumConcurrentLaneCount == 5) + #expect(MobileHostIrohApplicationLaneRouter.maximumConcurrentSimulatorStreamLaneCount == 2) + #expect( + MobileHostIrohApplicationLaneRouter.maximumConcurrentLaneCount + == MobileHostIrohApplicationLaneRouter.maximumConcurrentTerminalLaneCount + + MobileHostIrohApplicationLaneRouter.maximumConcurrentArtifactLaneCount + + MobileHostIrohApplicationLaneRouter.maximumConcurrentSimulatorStreamLaneCount + ) var quota = MobileHostIrohApplicationLaneQuota() let terminalIDs = (0..<5).map { _ in UUID() } @@ -582,8 +587,18 @@ extension MobileHostAuthorizationTests { #expect(didReserveArtifact) let didReserveSecondArtifact = quota.reserve(UUID(), laneClass: .artifact) #expect(!didReserveSecondArtifact) + let simulatorStreamIDs = (0..<3).map { _ in UUID() } + for id in simulatorStreamIDs.prefix(2) { + let didReserve = quota.reserve(id, laneClass: .simulatorStream) + #expect(didReserve) + } + let didReserveThirdSimulatorStream = quota.reserve( + simulatorStreamIDs[2], laneClass: .simulatorStream + ) + #expect(!didReserveThirdSimulatorStream) #expect(quota.terminalCount == 4) #expect(quota.artifactCount == 1) + #expect(quota.simulatorStreamCount == 2) quota.release(terminalIDs[0]) let didReuseTerminalCredit = quota.reserve(terminalIDs[4], laneClass: .terminal) @@ -591,6 +606,11 @@ extension MobileHostAuthorizationTests { quota.release(artifactID) let didReuseArtifactCredit = quota.reserve(UUID(), laneClass: .artifact) #expect(didReuseArtifactCredit) + quota.release(simulatorStreamIDs[0]) + let didReuseSimulatorStreamCredit = quota.reserve( + simulatorStreamIDs[2], laneClass: .simulatorStream + ) + #expect(didReuseSimulatorStreamCredit) } private func irohPeer( diff --git a/cmuxTests/MobileHostOrderedInputTests.swift b/cmuxTests/MobileHostOrderedInputTests.swift index 0f6147290b98..4038c55a946b 100644 --- a/cmuxTests/MobileHostOrderedInputTests.swift +++ b/cmuxTests/MobileHostOrderedInputTests.swift @@ -75,9 +75,13 @@ struct MobileHostOrderedInputTests { try Self.framedBatch([("input-overflow", "terminal.input")]) ) - #expect(await transport.closeCount() == 1) + let responses = await transport.waitForResponseCount(1) + #expect(responses == ["input-overflow"]) + #expect(await transport.errorCode(for: "input-overflow") == "server_busy") + #expect(await transport.closeCount() == 0) #expect(await gate.handledRequestCount() == 1) await gate.releaseFirstInput() + await connection.close(reason: "test complete") } @Test @@ -256,6 +260,7 @@ private actor OrderedInputHandlerGate { private actor OrderedInputRecordingTransport: CmxByteTransport { private var responses: [String] = [] + private var errorCodes: [String: String] = [:] private var responseWaiters: [(Int, CheckedContinuation<[String], Never>)] = [] private var closes = 0 @@ -286,7 +291,9 @@ private actor OrderedInputRecordingTransport: CmxByteTransport { for payload in payloads { let envelope = try JSONSerialization.jsonObject(with: payload) as? [String: Any] - responses.append(envelope?["id"] as? String ?? "") + let id = envelope?["id"] as? String ?? "" + responses.append(id) + errorCodes[id] = (envelope?["error"] as? [String: Any])?["code"] as? String } let ready = responseWaiters.filter { responses.count >= $0.0 } responseWaiters.removeAll { responses.count >= $0.0 } @@ -306,6 +313,7 @@ private actor OrderedInputRecordingTransport: CmxByteTransport { } } + func errorCode(for id: String) -> String? { errorCodes[id] } func responseIDs() -> [String] { responses } func closeCount() -> Int { closes } }