Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable {
/// - ticket: The attach ticket authorizing requests.
/// - allowsStackAuthFallback: When `true`, falls back to a Stack Auth token
/// on routes that allow it once the attach ticket no longer covers a request.
/// - transportConnectObserver: Optional observer for the underlying transport
/// attempt, success, and failure lifecycle.
public init(
runtime: any MobileSyncRuntime,
route: CmxAttachRoute,
Expand All @@ -40,7 +42,8 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable {
stackTokenForceRefreshGate: RPCStackTokenGate? = nil,
abandonedConnectCleanupTimeoutNanoseconds: UInt64 = 1_000_000_000,
lateAbandonedConnectCloseTimeoutNanoseconds: UInt64 = 5_000_000_000,
stackTokenGateResetNanoseconds: UInt64 = 30_000_000_000
stackTokenGateResetNanoseconds: UInt64 = 30_000_000_000,
transportConnectObserver: (@Sendable (MobileRPCTransportConnectEvent) async -> Void)? = nil
) {
self.runtime = runtime
self.route = route
Expand Down Expand Up @@ -73,7 +76,8 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable {
makeTransport: { [runtime, transportRequest] in
try runtime.transportFactory.makeTransport(for: transportRequest)
},
makeIndependentEventByteStream: independentEventFactory
makeIndependentEventByteStream: independentEventFactory,
transportConnectObserver: transportConnectObserver
)
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ actor MobileCoreRPCSession {
typealias TransportFactory = @Sendable () throws -> any CmxByteTransport
typealias IndependentEventByteStreamFactory = @Sendable () async throws -> CmxIndependentEventByteStream
typealias ConnectedCandidateHook = @Sendable (_ candidate: any CmxByteTransport) async -> Void
typealias TransportConnectObserver = @Sendable (MobileRPCTransportConnectEvent) async -> Void
typealias PendingContinuation = CheckedContinuation<Result<Data, MobileShellConnectionError>, Never>
typealias ConnectingTask = (id: UUID, lease: MobileRPCConnectAttemptLease?, task: Task<any CmxByteTransport, any Error>, waiters: Set<UUID>, completed: Bool)
static let defaultAbandonedConnectCleanupTimeoutNanoseconds: UInt64 = 1_000_000_000
Expand Down Expand Up @@ -48,6 +49,7 @@ actor MobileCoreRPCSession {
private let makeTransport: TransportFactory
let makeIndependentEventByteStream: IndependentEventByteStreamFactory?
private let didReceiveConnectedCandidate: ConnectedCandidateHook?
private let transportConnectObserver: TransportConnectObserver?
private var transport: (any CmxByteTransport)?
private var connectionTask: ConnectingTask?
private var installedConnectionID: UUID?
Expand All @@ -73,7 +75,8 @@ actor MobileCoreRPCSession {
lateAbandonedConnectCloseTimeoutNanoseconds: UInt64 = 5_000_000_000,
makeTransport: @escaping TransportFactory,
makeIndependentEventByteStream: IndependentEventByteStreamFactory? = nil,
didReceiveConnectedCandidate: ConnectedCandidateHook? = nil
didReceiveConnectedCandidate: ConnectedCandidateHook? = nil,
transportConnectObserver: TransportConnectObserver? = nil
) {
self.connectAttemptKey = connectAttemptKey
self.connectAttemptRegistry = connectAttemptRegistry
Expand All @@ -82,6 +85,7 @@ actor MobileCoreRPCSession {
self.makeTransport = makeTransport
self.makeIndependentEventByteStream = makeIndependentEventByteStream
self.didReceiveConnectedCandidate = didReceiveConnectedCandidate
self.transportConnectObserver = transportConnectObserver
}

deinit {
Expand Down Expand Up @@ -228,22 +232,40 @@ actor MobileCoreRPCSession {
} else {
connectLease = .untracked
}
let candidate: any CmxByteTransport
do {
candidate = try makeTransport()
} catch {
await connectAttemptRegistry.clearFinishedConnect(lease: connectLease)
throw error
}
let connectStartedAt = ContinuousClock.now
let makeTransport = makeTransport
let transportConnectObserver = transportConnectObserver
connectionID = UUID()
task = Task.detached {
try await withTaskCancellationHandler {
try await candidate.connect()
return candidate
} onCancel: {
Task {
await candidate.close()
await transportConnectObserver?(.attempt)
do {
try Task.checkCancellation()
let candidate = try makeTransport()
let connected = try await withTaskCancellationHandler {
try await candidate.connect()
return candidate
} onCancel: {
Task {
await candidate.close()
}
}
await transportConnectObserver?(
.connected(elapsedMilliseconds: Self.elapsedMilliseconds(since: connectStartedAt))
)
return connected
} catch is CancellationError {
throw CancellationError()
} catch {
if Task.isCancelled {
throw CancellationError()
}
await transportConnectObserver?(
.failed(
error: error,
elapsedMilliseconds: Self.elapsedMilliseconds(since: connectStartedAt)
)
)
throw error
}
}
Comment on lines +258 to 270

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Cancellation reported as transport failure in dial diagnostics

When tearDown cancels the in-flight connection task, Task.checkCancellation() or candidate.connect() throws CancellationError, which falls into the catch block and calls transportConnectObserver?(.failed(error: CancellationError, ...)). At the consumer side, mobileDialFailureKind has no CancellationError branch, so every intentional cancellation is logged as failure_kind=transport_error:CancellationError. This inflates observed transport-failure counts in the dial log whenever the user navigates away or a session tears down mid-connect. The fix is to rethrow without calling the observer when the error is a cancellation, so .attempt without a corresponding .connected or .failed implicitly signals that the attempt was abandoned.

Suggested change
} catch {
await transportConnectObserver?(
.failed(
error: error,
elapsedMilliseconds: Self.elapsedMilliseconds(since: connectStartedAt)
)
)
throw error
}
}
} catch {
if !(error is CancellationError) {
await transportConnectObserver?(
.failed(
error: error,
elapsedMilliseconds: Self.elapsedMilliseconds(since: connectStartedAt)
)
)
}
throw error
}

connectionTask = (id: connectionID, lease: connectLease, task: task, waiters: [waiterID], completed: false)
Expand Down Expand Up @@ -325,6 +347,13 @@ actor MobileCoreRPCSession {
return candidate
}

private static func elapsedMilliseconds(since start: ContinuousClock.Instant) -> Int {
let components = start.duration(to: .now).components
let milliseconds = components.seconds * 1_000
+ components.attoseconds / 1_000_000_000_000_000
return max(0, Int(milliseconds))
}

private func cancelConnectingWaiter(id connectionID: UUID, waiterID: UUID) async {
guard transport == nil, connectionTask?.id == connectionID, let task = connectionTask?.task else {
return
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
import Foundation

/// A lifecycle event for one underlying mobile transport connection attempt.
public enum MobileRPCTransportConnectEvent: Sendable {
/// The transport factory is about to build and dial its route.
case attempt
/// The underlying byte transport connected successfully.
/// - Parameter elapsedMilliseconds: Whole milliseconds since the attempt began.
case connected(elapsedMilliseconds: Int)
/// The transport factory or underlying byte transport failed.
/// - Parameters:
/// - error: The transport construction or connection error.
/// - elapsedMilliseconds: Whole milliseconds since the attempt began.
case failed(error: any Error, elapsedMilliseconds: Int)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
import CMUXMobileCore
import Foundation
@testable import CmuxMobileRPC

actor FirstConnectClosedErrorThenSucceedsTransport: CmxByteTransport {
private struct SyntheticClosedError: Error {}

private var sentPayloads: [Data] = []
private var receiveWaiters: [CheckedContinuation<Data?, Never>] = []
private var queuedResponses: [Data] = []
private var firstConnectRelease: CheckedContinuation<Void, Never>?
private var firstConnectStartedWaiters: [CheckedContinuation<Void, Never>] = []
private var firstConnectFinishedWaiters: [CheckedContinuation<Void, Never>] = []
private var firstConnectStarted = false
private var firstConnectFinished = false
private var connects = 0
private var isClosed = false

func connect() async throws {
connects += 1
if connects == 1 {
firstConnectStarted = true
let startedWaiters = firstConnectStartedWaiters
firstConnectStartedWaiters = []
for waiter in startedWaiters {
waiter.resume()
}
await withCheckedContinuation { continuation in
firstConnectRelease = continuation
}
firstConnectFinished = true
let finishedWaiters = firstConnectFinishedWaiters
firstConnectFinishedWaiters = []
for waiter in finishedWaiters {
waiter.resume()
}
throw SyntheticClosedError()
}
isClosed = false
}

func receive() async throws -> Data? {
if isClosed {
return nil
}
if !queuedResponses.isEmpty {
return queuedResponses.removeFirst()
}
return await withCheckedContinuation { continuation in
receiveWaiters.append(continuation)
}
}

func send(_ data: Data) async throws {
var buffer = data
let payloads = try MobileSyncFrameCodec.decodeFrames(from: &buffer)
sentPayloads.append(contentsOf: payloads)
for payload in payloads {
let request = try recordedRPCRequest(from: payload)
try enqueueResponse(id: request.id)
}
}

func close() async {
isClosed = true
if connects == 1 {
firstConnectRelease?.resume()
firstConnectRelease = nil
return
}
let waiters = receiveWaiters
receiveWaiters = []
for waiter in waiters {
waiter.resume(returning: nil)
}
}

func waitUntilFirstConnectStarted() async {
if firstConnectStarted {
return
}
await withCheckedContinuation { continuation in
firstConnectStartedWaiters.append(continuation)
}
}

func waitUntilFirstConnectFinished() async {
if firstConnectFinished {
return
}
await withCheckedContinuation { continuation in
firstConnectFinishedWaiters.append(continuation)
}
}

func connectCount() -> Int {
connects
}

func sentRequests() throws -> [RecordedRPCRequest] {
try sentPayloads.map(recordedRPCRequest(from:))
}

private func enqueueResponse(id: String?) throws {
let response: [String: Any] = [
"id": id ?? "",
"ok": true,
"result": ["status": "ok"],
]
let payload = try JSONSerialization.data(withJSONObject: response)
let frame = try MobileSyncFrameCodec.encodeFrame(payload)
if let waiter = receiveWaiters.first {
receiveWaiters.removeFirst()
waiter.resume(returning: frame)
} else {
queuedResponses.append(frame)
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -132,11 +132,16 @@ import Testing
expiresAt: Date().addingTimeInterval(60),
authToken: "ticket-secret"
)
let (connectEvents, connectEventContinuation) =
AsyncStream<MobileRPCTransportConnectEvent>.makeStream()
let client = MobileCoreRPCClient(
runtime: runtime,
route: route,
ticket: ticket,
allowsStackAuthFallback: true
allowsStackAuthFallback: true,
transportConnectObserver: { event in
connectEventContinuation.yield(event)
}
)
let first = try MobileCoreRPCClient.requestData(
method: "terminal.input",
Expand Down Expand Up @@ -164,6 +169,18 @@ import Testing
} catch {
Issue.record("Expected CancellationError, got \(error)")
}
connectEventContinuation.finish()
var recordedConnectEvents: [MobileRPCTransportConnectEvent] = []
for await event in connectEvents {
recordedConnectEvents.append(event)
}
#expect(recordedConnectEvents.count == 1)
if let firstEvent = recordedConnectEvents.first {
guard case .attempt = firstEvent else {
Issue.record("Expected only an attempt event for a cancelled connect")
return
}
}

let data = try await client.sendRequest(second)
let response = try #require(JSONSerialization.jsonObject(with: data) as? [String: String])
Expand All @@ -172,6 +189,75 @@ import Testing
#expect(try await transport.sentRequests().map(\.id) == ["second-after-connect-cancellation"])
}

@Test func callerCancellationClosedErrorDoesNotEmitFailedAndAllowsRetry() async throws {
let transport = FirstConnectClosedErrorThenSucceedsTransport()
let (connectEvents, connectEventContinuation) =
AsyncStream<MobileRPCTransportConnectEvent>.makeStream()
let session = MobileCoreRPCSession(
makeTransport: { transport },
transportConnectObserver: { event in
connectEventContinuation.yield(event)
}
)
let first = try MobileCoreRPCClient.requestData(
method: "mobile.host.status",
params: [:],
id: "cancelled-closed-connect"
)
let second = try MobileCoreRPCClient.requestData(
method: "mobile.host.status",
params: [:],
id: "retry-after-closed-connect"
)
let deadline = DispatchTime.now().uptimeNanoseconds + 60 * 1_000_000_000
let firstTask = Task {
try await session.send(
payload: first,
requestID: "cancelled-closed-connect",
deadlineUptimeNanoseconds: deadline
)
}

await transport.waitUntilFirstConnectStarted()
firstTask.cancel()
do {
_ = try await firstTask.value
Issue.record("Expected first RPC request to throw CancellationError")
} catch is CancellationError {
} catch {
Issue.record("Expected CancellationError, got \(error)")
}
await transport.waitUntilFirstConnectFinished()

let data = try await session.send(
payload: second,
requestID: "retry-after-closed-connect",
deadlineUptimeNanoseconds: deadline
)
let response = try #require(JSONSerialization.jsonObject(with: data) as? [String: String])
#expect(response["status"] == "ok")
#expect(await transport.connectCount() == 2)
#expect(try await transport.sentRequests().map(\.id) == ["retry-after-closed-connect"])

connectEventContinuation.finish()
var attemptCount = 0
var connectedCount = 0
var failedCount = 0
for await event in connectEvents {
switch event {
case .attempt:
attemptCount += 1
case .connected:
connectedCount += 1
case .failed:
failedCount += 1
}
}
#expect(attemptCount == 2)
#expect(connectedCount == 1)
#expect(failedCount == 0)
}

@Test func repeatedConnectTimeoutsDoNotFanOutWhileCleanupIsStuck() async throws {
let transport = CancellationIgnoringConnectTransport()
let route = try hostPortRoute(kind: .debugLoopback, host: "127.0.0.1", port: 59127)
Expand Down
Loading