Skip to content
Merged
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 @@ -11,14 +11,18 @@ public struct CmxByteTransportRequest: Equatable, Sendable {
public let route: CmxAttachRoute
public let expectedPeerDeviceID: String?
public let authorizationMode: CmxTransportAuthorizationMode
/// The local owner whose network path this request represents.
public let sessionPurpose: CmxTransportSessionPurpose

public init(
route: CmxAttachRoute,
expectedPeerDeviceID: String?,
authorizationMode: CmxTransportAuthorizationMode
authorizationMode: CmxTransportAuthorizationMode,
sessionPurpose: CmxTransportSessionPurpose = .foregroundControl
) {
self.route = route
self.expectedPeerDeviceID = expectedPeerDeviceID
self.authorizationMode = authorizationMode
self.sessionPurpose = sessionPurpose
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
/// The local lifecycle role of a byte-transport request.
///
/// This value never crosses the network. Transport pools use it to keep
/// foreground diagnostics anchored to the user-visible RPC connection when
/// background aggregation or feature lanes share the same endpoint.
public enum CmxTransportSessionPurpose: UInt8, Equatable, Sendable {
/// The control session powering the currently visible Mac workspace.
case foregroundControl = 1
/// A control session keeping a non-selected Mac's workspace list current.
case backgroundControl = 2
/// A short-lived request that discovers or validates a route.
case probe = 3
/// An independent feature lane sharing an admitted Iroh session.
case featureLane = 4
}
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,8 @@ public enum DiagnosticEventCode: UInt16, Sendable, Codable, CaseIterable {
/// A relay-policy refresh failed. `b`, when present, is
/// ``DiagnosticFailureKind``.
case relayPolicyRefreshFailed = 39
/// The selected network path changed. `a` is ``DiagnosticPathKind``.
/// The selected network path changed. `a` is ``DiagnosticPathKind``. The
/// foreground control session wins over background and feature sessions.
case selectedPathChanged = 40
/// An established app-transport session closed. `b`, when present, is
/// ``DiagnosticFailureKind``; absence means an expected closure.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,15 @@ actor CmxIrohClientSessionPool {
private struct ControlWaiter {
let id: UUID
let ownerID: UUID
let purpose: CmxTransportSessionPurpose
let continuation: CheckedContinuation<Void, Never>
}

private struct ControlOwner {
let id: UUID
let purpose: CmxTransportSessionPurpose
}

private let supervisor: CmxIrohEndpointSupervisor
private let contextProvider: any CmxIrohClientContextProvider
private let protocolConfiguration: CmxIrohProtocolConfiguration
Expand All @@ -35,7 +41,7 @@ actor CmxIrohClientSessionPool {
private var sessions: [SessionKey: PooledSession] = [:]
private var sessionOrder: [SessionKey] = []
private var connectionTasks: [SessionKey: PendingConnection] = [:]
private var controlOwners: [SessionKey: UUID] = [:]
private var controlOwners: [SessionKey: ControlOwner] = [:]
private var controlWaiters: [SessionKey: [ControlWaiter]] = [:]
private var selectedPathContinuations: [UUID: AsyncStream<Void>.Continuation] = [:]

Expand Down Expand Up @@ -174,14 +180,18 @@ actor CmxIrohClientSessionPool {
ownerID: UUID
) async throws -> CmxIrohClientSession {
let key = try sessionKey(for: request)
try await reserveControlOwner(for: key, ownerID: ownerID)
try await reserveControlOwner(
for: key,
ownerID: ownerID,
purpose: request.sessionPurpose
)
do {
return try await session(
for: request,
preservesControlOwnerOnClosed: true
)
} catch {
if controlOwners[key] == ownerID {
if controlOwners[key]?.id == ownerID {
releaseControlOwner(for: key, ownerID: ownerID)
}
throw error
Expand Down Expand Up @@ -223,7 +233,7 @@ actor CmxIrohClientSessionPool {
ownerID: UUID
) async {
guard let key = try? sessionKey(for: request),
controlOwners[key] == ownerID else {
controlOwners[key]?.id == ownerID else {
return
}
await invalidateSession(for: key, releasesControlOwner: false)
Expand Down Expand Up @@ -256,7 +266,14 @@ actor CmxIrohClientSessionPool {
}

func selectedObservedPath() async -> CmxIrohObservedConnectionPath {
guard let key = sessionOrder.last,
let foregroundKey = sessionOrder.last { key in
controlOwners[key]?.purpose == .foregroundControl
&& sessions[key] != nil
}
let controlKey = sessionOrder.last { key in
controlOwners[key] != nil && sessions[key] != nil
}
guard let key = foregroundKey ?? controlKey ?? sessionOrder.last,
let session = sessions[key]?.session else { return .unavailable }
return await session.observedSelectedPath()
}
Expand All @@ -279,13 +296,13 @@ actor CmxIrohClientSessionPool {

private func sessionDidClose(key: SessionKey, sessionID: UUID) async {
guard let pooled = sessions[key], pooled.id == sessionID else { return }
let ownerID = controlOwners[key]
let owner = controlOwners[key]
sessions[key] = nil
sessionOrder.removeAll { $0 == key }
pooled.pathObservationTask.cancel()
await pooled.session.close()
if let ownerID {
releaseControlOwner(for: key, ownerID: ownerID)
if let owner {
releaseControlOwner(for: key, ownerID: owner.id)
}
publishSelectedPathChange()
}
Expand All @@ -307,16 +324,16 @@ actor CmxIrohClientSessionPool {
releasesControlOwner: Bool = true
) async {
if let expectedID, sessions[key]?.id != expectedID { return }
let ownerID = releasesControlOwner ? controlOwners[key] : nil
let owner = releasesControlOwner ? controlOwners[key] : nil
connectionTasks[key]?.task.cancel()
connectionTasks[key] = nil
let pooled = sessions.removeValue(forKey: key)
sessionOrder.removeAll { $0 == key }
pooled?.closureTask.cancel()
pooled?.pathObservationTask.cancel()
await pooled?.session.close()
if let ownerID {
releaseControlOwner(for: key, ownerID: ownerID)
if let owner {
releaseControlOwner(for: key, ownerID: owner.id)
}
publishSelectedPathChange()
}
Expand All @@ -331,12 +348,14 @@ actor CmxIrohClientSessionPool {

private func reserveControlOwner(
for key: SessionKey,
ownerID: UUID
ownerID: UUID,
purpose: CmxTransportSessionPurpose
) async throws {
if let existing = controlOwners[key] {
if existing == ownerID { return }
if existing.id == ownerID { return }
} else {
controlOwners[key] = ownerID
controlOwners[key] = ControlOwner(id: ownerID, purpose: purpose)
publishSelectedPathChangeIfEstablished(for: key)
return
}

Expand All @@ -348,17 +367,19 @@ actor CmxIrohClientSessionPool {
return
}
if let existing = controlOwners[key] {
if existing == ownerID {
if existing.id == ownerID {
continuation.resume()
} else {
controlWaiters[key, default: []].append(ControlWaiter(
id: waiterID,
ownerID: ownerID,
purpose: purpose,
continuation: continuation
))
}
} else {
controlOwners[key] = ownerID
controlOwners[key] = ControlOwner(id: ownerID, purpose: purpose)
publishSelectedPathChangeIfEstablished(for: key)
continuation.resume()
}
Comment on lines 349 to 384

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 publishSelectedPathChange() fires before the session is established

publishSelectedPathChange() is now called inside reserveControlOwner at both the fast-path (controlOwners[key] = ...; publishSelectedPathChange(); return) and the slow-path continuation branch. At that point sessions[key] is still nil — the session hasn't been established yet. selectedObservedPath() guards foregroundKey and controlKey with sessions[key] != nil, so the notification causes consumers to wake and read the old fallback path (or .unavailable) rather than the new foreground session's path. The session establishment fires a second notification that brings the correct value, so state is eventually consistent. The concern is that consumers will briefly observe either stale data from an older background session or .unavailable between control-owner registration and session connect, which could cause a visible flicker in connection-status UI. Consider deferring these two publishSelectedPathChange() calls to the moment the session is added to sessions (which already fires its own notification via the session(for:) path).

}
Expand All @@ -368,12 +389,12 @@ actor CmxIrohClientSessionPool {

do {
try Task.checkCancellation()
guard controlOwners[key] == ownerID else {
guard controlOwners[key]?.id == ownerID else {
throw CmxIrohClientRuntimeError.inactive
}
} catch {
cancelControlWaiter(for: key, id: waiterID)
if controlOwners[key] == ownerID {
if controlOwners[key]?.id == ownerID {
releaseControlOwner(for: key, ownerID: ownerID)
}
throw error
Expand All @@ -391,15 +412,24 @@ actor CmxIrohClientSessionPool {
}

private func releaseControlOwner(for key: SessionKey, ownerID: UUID) {
guard controlOwners[key] == ownerID else { return }
guard controlOwners[key]?.id == ownerID else { return }
controlOwners[key] = nil
guard var waiters = controlWaiters[key], !waiters.isEmpty else { return }
guard var waiters = controlWaiters[key], !waiters.isEmpty else {
publishSelectedPathChangeIfEstablished(for: key)
return
}
let next = waiters.removeFirst()
controlWaiters[key] = waiters.isEmpty ? nil : waiters
controlOwners[key] = next.ownerID
controlOwners[key] = ControlOwner(id: next.ownerID, purpose: next.purpose)
publishSelectedPathChangeIfEstablished(for: key)
next.continuation.resume()
}

private func publishSelectedPathChangeIfEstablished(for key: SessionKey) {
guard sessions[key] != nil else { return }
publishSelectedPathChange()
}

private func publishSelectedPathChange() {
for continuation in selectedPathContinuations.values {
continuation.yield(())
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -578,6 +578,91 @@ struct CmxIrohClientSessionPoolTests {
#expect(await iterator.next() != nil)
#expect(await pool.selectedObservedPath() == .unavailable)
}

@Test
func selectedPathDoesNotPublishAnUnestablishedControlSession() async throws {
let fixture = try PoolFixture()
let endpoint = TestHangingDialEndpoint(localIdentity: fixture.localIdentity)
let pool = try await fixture.pool(endpoint: endpoint, generation: 1)
let changes = await pool.selectedPathChanges()
let recorder = SelectedPathChangeRecorder()
let observation = Task {
for await _ in changes {
await recorder.record()
}
}
#expect(await waitForSelectedPathChangeCount(recorder, atLeast: 1))

let transport = try CmxIrohByteTransportFactory(sessionPool: pool)
.makeTransport(for: fixture.request)
let connection = Task {
try await transport.connect()
}
let started = await endpoint.startedEvents()
var startedIterator = started.makeAsyncIterator()
#expect(await startedIterator.next() != nil)

let publishedBeforeEstablishment = await waitForSelectedPathChangeCount(
recorder,
atLeast: 2
)
#expect(!publishedBeforeEstablishment)
#expect(await pool.selectedObservedPath() == .unavailable)

connection.cancel()
await pool.deactivate()
_ = try? await connection.value
observation.cancel()
}

@Test
func selectedPathPrefersTheActiveControlSessionOverANewerBackgroundSession() async throws {
let fixture = try PoolFixture()
let backgroundIdentity = try CmxIrohPeerIdentity(
endpointID: String(repeating: "ef", count: 32)
)
let backgroundRequest = CmxByteTransportRequest(
route: try CmxAttachRoute(
id: "iroh-background-session",
kind: .iroh,
endpoint: .peer(identity: backgroundIdentity, pathHints: [])
),
expectedPeerDeviceID: "123e4567-e89b-42d3-a456-426614174031",
authorizationMode: .transportAdmission,
sessionPurpose: .backgroundControl
)
let controlConnection = TestIrohConnection(
remoteIdentity: fixture.remoteIdentity,
bidirectionalStreams: [fixture.controlStream()],
selectedPath: .direct
)
let backgroundConnection = TestIrohConnection(
remoteIdentity: backgroundIdentity,
bidirectionalStreams: [fixture.controlStream()],
selectedPath: .relay(url: "https://relay.example.com/")
)
let endpoint = TestDialingIrohEndpoint(
localIdentity: fixture.localIdentity,
dialResults: [
.connection(controlConnection),
.connection(backgroundConnection),
]
)
let pool = try await fixture.pool(endpoint: endpoint, generation: 1)
let control = try CmxIrohByteTransportFactory(sessionPool: pool)
.makeTransport(for: fixture.request)
let background = try CmxIrohByteTransportFactory(sessionPool: pool)
.makeTransport(for: backgroundRequest)

try await control.connect()
#expect(await pool.selectedObservedPath() == .direct)

try await background.connect()

#expect(await pool.selectedObservedPath() == .direct)
await background.close()
await control.close()
}
}

private func waitForControlWaiter(
Expand All @@ -591,6 +676,29 @@ private func waitForControlWaiter(
return false
}

private actor SelectedPathChangeRecorder {
private var count = 0

func record() {
count += 1
}

func observedCount() -> Int {
count
}
}

private func waitForSelectedPathChangeCount(
_ recorder: SelectedPathChangeRecorder,
atLeast expectedCount: Int
) async -> Bool {
for _ in 0 ..< 1_000 {
if await recorder.observedCount() >= expectedCount { return true }
await Task.yield()
}
return false
}

private struct PoolFixture {
let localIdentity: CmxIrohPeerIdentity
let remoteIdentity: CmxIrohPeerIdentity
Expand Down Expand Up @@ -620,7 +728,7 @@ private struct PoolFixture {
}

func pool(
endpoint: TestDialingIrohEndpoint,
endpoint: any CmxIrohEndpoint,
generation: UInt64,
contextProvider: (any CmxIrohClientContextProvider)? = nil
) async throws -> CmxIrohClientSessionPool {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,10 @@ actor TestHangingDialEndpoint: CmxIrohEndpoint {
to _: CmxIrohEndpointAddress,
alpn _: Data
) async throws -> any CmxIrohConnection {
startedContinuation.yield()
return try await withTaskCancellationHandler(operation: {
try await withCheckedThrowingContinuation { continuation in
pendingConnect = continuation
startedContinuation.yield()
}
}, onCancel: { [cancelledContinuation] in
cancelledContinuation.yield()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,15 +44,17 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable {
abandonedConnectCleanupTimeoutNanoseconds: UInt64 = 1_000_000_000,
lateAbandonedConnectCloseTimeoutNanoseconds: UInt64 = 5_000_000_000,
stackTokenGateResetNanoseconds: UInt64 = 30_000_000_000,
transportConnectObserver: (@Sendable (MobileRPCTransportConnectEvent) -> Void)? = nil
transportConnectObserver: (@Sendable (MobileRPCTransportConnectEvent) -> Void)? = nil,
sessionPurpose: CmxTransportSessionPurpose = .foregroundControl
) {
self.runtime = runtime
self.route = route
self.ticket = ticket
let transportRequest = CmxByteTransportRequest(
route: route,
expectedPeerDeviceID: ticket.macDeviceID,
authorizationMode: route.kind == .iroh ? .transportAdmission : .stackBearer
authorizationMode: route.kind == .iroh ? .transportAdmission : .stackBearer,
sessionPurpose: sessionPurpose
)
self.transportRequest = transportRequest
self.allowsStackAuthFallback = allowsStackAuthFallback
Expand Down
Loading