diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxEventLaneIO.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxEventLaneIO.swift new file mode 100644 index 000000000000..2951fb46c113 --- /dev/null +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxEventLaneIO.swift @@ -0,0 +1,104 @@ +public import Foundation + +/// Write half of one server->client event lane. ``IrxStreamWriter`` is the +/// production conformer; tests substitute in-memory lanes. +public protocol IrxEventLaneWriting: Sendable { + func write(_ data: Data) async throws + func setPriority(_ priority: Int32) async throws + func finish() async + func reset(errorCode: UInt64) async +} + +/// Read half of one server->client event lane. ``IrxStreamReader`` is the +/// production conformer; tests substitute in-memory lanes. +public protocol IrxEventLaneReading: Sendable { + /// Whatever bytes are available, or nil on EOF. + func readRaw() async throws -> Data? + func stop(errorCode: UInt64) async +} + +extension IrxStreamWriter: IrxEventLaneWriting {} +extension IrxStreamReader: IrxEventLaneReading {} + +/// Wire vocabulary for per-surface event lanes. +/// +/// A surface lane is an `events` lane whose descriptor names the terminal it +/// carries (`terminal:`). The shared events lane has no resource, +/// so a peer that predates surface lanes never confuses the two. The Mac only +/// opens surface lanes after the phone advertised ``subscribeParameterValue`` +/// in `mobile.events.subscribe`, because an older phone accepts exactly one +/// uni stream and would never read a second one. +public struct IrxSurfaceEventLaneProtocol: Sendable { + public init() {} + + /// `mobile.events.subscribe` parameter a client sends to opt in, and the + /// acknowledgement key a host echoes when it granted surface lanes. + public let subscribeParameterKey = "surface_event_lanes" + public let subscribeParameterValue = "v1" + let resourcePrefix = "terminal:" + + public func descriptor(surfaceID: String) -> IrxLaneDescriptor { + IrxLaneDescriptor( + lane: .events, + resource: resourcePrefix + normalizedSurfaceID(surfaceID) + ) + } + + /// The surface a lane carries, or nil for the shared events lane. + public func surfaceID(of descriptor: IrxLaneDescriptor) -> String? { + guard descriptor.lane == .events, + let resource = descriptor.resource, + resource.hasPrefix(resourcePrefix) else { return nil } + let surfaceID = String(resource.dropFirst(resourcePrefix.count)) + return surfaceID.isEmpty ? nil : surfaceID + } + + public func normalizedSurfaceID(_ surfaceID: String) -> String { + surfaceID.trimmingCharacters(in: .whitespacesAndNewlines).lowercased() + } +} + +/// Splits a lane's byte stream at mobile-sync frame boundaries (4-byte +/// big-endian length + payload). Readers that merge several lanes into one +/// consumer must only forward whole frames, or bytes from two lanes would +/// interleave inside a frame. +struct IrxEventFrameAligner: Sendable { + enum Failure: Error, Equatable { + case frameTooLarge(Int) + } + + static let headerByteCount = 4 + + let maximumFrameByteCount: Int + private var buffer = Data() + + init(maximumFrameByteCount: Int) { + self.maximumFrameByteCount = maximumFrameByteCount + } + + var hasPartialFrame: Bool { !buffer.isEmpty } + + /// Appends `chunk` and returns every complete frame now available, still + /// framed, as one contiguous block; nil when no frame completed. + mutating func append(_ chunk: Data) throws -> Data? { + buffer.append(chunk) + var consumed = 0 + while buffer.count - consumed >= Self.headerByteCount { + let start = buffer.startIndex + consumed + var length = 0 + for byte in buffer[start..<(start + Self.headerByteCount)] { + length = (length << 8) | Int(byte) + } + guard length <= maximumFrameByteCount else { + throw Failure.frameTooLarge(length) + } + let frameByteCount = Self.headerByteCount + length + guard buffer.count - consumed >= frameByteCount else { break } + consumed += frameByteCount + } + guard consumed > 0 else { return nil } + let complete = Data(buffer.prefix(consumed)) + buffer.removeFirst(consumed) + return complete + } +} diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxServerEventLaneHub.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxServerEventLaneHub.swift new file mode 100644 index 000000000000..2b408ddc95b1 --- /dev/null +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxServerEventLaneHub.swift @@ -0,0 +1,192 @@ +public import Foundation + +/// Client-side owner of every server->client event lane on one irx +/// connection: the shared events lane plus one lane per terminal surface. +/// +/// Each lane has its own reader task, so a burst on one surface's stream +/// never delays bytes already delivered on another stream. Readers forward +/// only whole mobile-sync frames into the single subscriber stream, which lets +/// the RPC session decode the merged output exactly like one ordered lane: +/// ordering holds per lane, and every surface's frames stay on one lane. +/// +/// The hub accepts lanes for the connection lifetime. Keeping one acceptor +/// per connection means a replaced subscriber can never leave a stale accept +/// loop behind that steals the next lane the host opens. +public actor IrxServerEventLaneHub { + public struct Limits: Sendable { + public var maximumSurfaceLaneCount: Int + public var maximumFrameByteCount: Int + + public init( + maximumSurfaceLaneCount: Int = 32, + maximumFrameByteCount: Int = 8 * 1024 * 1024 + ) { + self.maximumSurfaceLaneCount = max(1, maximumSurfaceLaneCount) + self.maximumFrameByteCount = maximumFrameByteCount + } + } + + public typealias AcceptLane = @Sendable () async throws -> (IrxLaneDescriptor, any IrxEventLaneReading)? + public typealias Output = AsyncThrowingStream + + // Stop codes the hub sends when it refuses a lane. + public static let unsupportedLaneStopCode: UInt64 = 2 + public static let laneLimitStopCode: UInt64 = 3 + public static let malformedFrameStopCode: UInt64 = 5 + + private let acceptLane: AcceptLane + private let limits: Limits + private let journal: IrxJournal? + private var acceptTask: Task? + private var readerTasks: [UInt64: Task] = [:] + private var readers: [UInt64: any IrxEventLaneReading] = [:] + private var surfaceLaneIDs: Set = [] + private var nextLaneID: UInt64 = 0 + private var subscriber: Output.Continuation? + private var subscriberID: UInt64 = 0 + private var terminalError: (any Error)? + private var isFinished = false + + public init(limits: Limits = Limits(), journal: IrxJournal? = nil, acceptLane: @escaping AcceptLane) { + self.acceptLane = acceptLane + self.limits = limits + self.journal = journal + } + + public var isAlive: Bool { !isFinished } + + public func activeSurfaceLaneCount() -> Int { surfaceLaneIDs.count } + + /// Returns the merged event byte stream, replacing any previous + /// subscriber (which finishes normally). The first call starts accepting. + public func subscribe() -> Output { + let (stream, continuation) = Output.makeStream() + guard !isFinished else { + continuation.finish(throwing: terminalError ?? CancellationError()) + return stream + } + subscriber?.finish() + subscriberID &+= 1 + let id = subscriberID + subscriber = continuation + continuation.onTermination = { [weak self] _ in + Task { await self?.subscriberEnded(id: id) } + } + if acceptTask == nil { + acceptTask = Task { [weak self] in await self?.runAcceptLoop() } + } + return stream + } + + /// Stops every lane. The native accept may stay blocked until the + /// connection closes; any lane it returns afterwards is stopped. + public func stop() async { + finish(error: nil) + let stopping = Array(readers.values) + readers.removeAll() + for reader in stopping { + await reader.stop(errorCode: 0) + } + } + + private func subscriberEnded(id: UInt64) { + guard subscriberID == id else { return } + subscriber = nil + } + + private func runAcceptLoop() async { + while !isFinished { + let accepted: (IrxLaneDescriptor, any IrxEventLaneReading)? + do { + accepted = try await acceptLane() + } catch { + finish(error: error) + return + } + guard let (descriptor, reader) = accepted else { + // Native connection closure ends lane acceptance. + finish(error: IrxConnectionError.closed(nil)) + return + } + await admit(descriptor: descriptor, reader: reader) + } + } + + private func admit(descriptor: IrxLaneDescriptor, reader: any IrxEventLaneReading) async { + guard !isFinished else { + await reader.stop(errorCode: 0) + return + } + guard descriptor.lane == .events else { + await reader.stop(errorCode: Self.unsupportedLaneStopCode) + return + } + let surfaceID = IrxSurfaceEventLaneProtocol().surfaceID(of: descriptor) + if surfaceID != nil, surfaceLaneIDs.count >= limits.maximumSurfaceLaneCount { + // The host treats a stopped lane as a failed write and falls back. + journal?.record("client-events", "surface-lane-refused", ["reason": "limit"]) + await reader.stop(errorCode: Self.laneLimitStopCode) + return + } + nextLaneID &+= 1 + let laneID = nextLaneID + readers[laneID] = reader + if surfaceID != nil { + surfaceLaneIDs.insert(laneID) + } + journal?.record( + "client-events", surfaceID == nil ? "lane-accepted" : "surface-lane-accepted", + ["open_surface_lanes": String(surfaceLaneIDs.count)] + ) + let maximumFrameByteCount = limits.maximumFrameByteCount + readerTasks[laneID] = Task { [weak self] in + var aligner = IrxEventFrameAligner(maximumFrameByteCount: maximumFrameByteCount) + var stopCode: UInt64? + do { + while !Task.isCancelled, let chunk = try await reader.readRaw() { + guard !chunk.isEmpty else { continue } + guard let frames = try aligner.append(chunk) else { continue } + await self?.deliver(frames) + } + } catch is IrxEventFrameAligner.Failure { + stopCode = Self.malformedFrameStopCode + } catch { + // A reset lane only loses its own unfinished frame. + } + if let stopCode { await reader.stop(errorCode: stopCode) } + await self?.laneEnded(laneID: laneID) + } + } + + private func deliver(_ frames: Data) { + subscriber?.yield(frames) + } + + private func laneEnded(laneID: UInt64) { + readerTasks.removeValue(forKey: laneID) + readers.removeValue(forKey: laneID) + if surfaceLaneIDs.remove(laneID) != nil { + journal?.record( + "client-events", "surface-lane-ended", + ["open_surface_lanes": String(surfaceLaneIDs.count)] + ) + } + } + + private func finish(error: (any Error)?) { + guard !isFinished else { return } + isFinished = true + terminalError = error + acceptTask?.cancel() + acceptTask = nil + for task in readerTasks.values { task.cancel() } + readerTasks.removeAll() + surfaceLaneIDs.removeAll() + if let error { + subscriber?.finish(throwing: error) + } else { + subscriber?.finish() + } + subscriber = nil + } +} diff --git a/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxSurfaceEventLanes.swift b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxSurfaceEventLanes.swift new file mode 100644 index 000000000000..a2c3cd12c1a2 --- /dev/null +++ b/Packages/Shared/CmuxIrxTransport/Sources/CmuxIrxTransport/IrxSurfaceEventLanes.swift @@ -0,0 +1,264 @@ +public import Foundation + +/// Server-side owner of one uni QUIC stream per terminal surface. +/// +/// QUIC delivers each stream in order, so a burst or replay for one terminal +/// on a shared stream head-of-line-blocks every other terminal's output, +/// including the echo of a key typed on another terminal. Each surface here +/// gets its own stream: a stalled or failed stream affects only its surface. +/// +/// Contract: +/// - A lane opens lazily on the first ``send(_:surfaceID:generation:)`` and is +/// keyed by `(surface, generation)`. A new generation finishes the previous +/// stream and opens a fresh one, so a caller that bumps the generation after +/// a failure never writes onto a stream whose earlier frames may be lost. +/// - A send that throws has already retired its lane (reset); the next send +/// reopens. Recovery is per surface and never touches the QUIC connection. +/// - A write that makes no progress within the stall deadline retires the lane +/// and throws ``LaneError/writeStalled``. The native reset may itself wait +/// for the stuck write (iroh-ffi serializes stream calls), so recovery never +/// waits for it: the next send opens a fresh stream while the reset runs +/// in the background. +/// - At most ``Configuration/maximumLaneCount`` lanes stay open; opening one +/// more finishes the least recently written lane. +/// - The focused surface's stream runs at ``Configuration/focusedPriority``; +/// every other surface shares ``Configuration/backgroundPriority`` with the +/// bulk events lane, so interactive echo is scheduled first. Priority +/// changes never wait on a lane's in-flight write, so noting focus from the +/// input path cannot delay input. +public actor IrxSurfaceEventLanes { + public struct Configuration: Sendable { + public var maximumLaneCount: Int + public var focusedPriority: Int32 + public var backgroundPriority: Int32 + public var openDeadline: Duration + public var stallDeadline: Duration + + public init( + maximumLaneCount: Int = 16, + focusedPriority: Int32 = 100, + backgroundPriority: Int32 = 50, + openDeadline: Duration = .seconds(5), + stallDeadline: Duration = .seconds(15) + ) { + self.maximumLaneCount = max(1, maximumLaneCount) + self.focusedPriority = focusedPriority + self.backgroundPriority = backgroundPriority + self.openDeadline = openDeadline + self.stallDeadline = stallDeadline + } + } + + public enum LaneError: Error, Equatable, Sendable { + case disabled + case openTimedOut + case writeStalled + } + + // Stream reset codes, visible to the phone as the lane's stop reason. + public static let supersededResetCode: UInt64 = 0 + public static let writeFailedResetCode: UInt64 = 6 + public static let stalledResetCode: UInt64 = 7 + + public typealias Opener = @Sendable (IrxLaneDescriptor) async throws -> any IrxEventLaneWriting + + private struct Lane { + let token: UInt64 + let generation: UInt64 + let writer: any IrxEventLaneWriting + var priority: Int32 + var lastUse: UInt64 + var priorityUpdate: Task? + } + + public nonisolated let configuration: Configuration + private let open: Opener + private let journal: IrxJournal? + private var lanes: [String: Lane] = [:] + private var focusedSurfaceID: String? + private var isEnabled = true + private var nextToken: UInt64 = 0 + private var useTick: UInt64 = 0 + + public init( + configuration: Configuration = Configuration(), + journal: IrxJournal? = nil, + open: @escaping Opener + ) { + self.configuration = configuration + self.journal = journal + self.open = open + } + + /// Writes one complete frame onto the surface's lane. + public func send(_ data: Data, surfaceID rawSurfaceID: String, generation: UInt64) async throws { + guard isEnabled else { throw LaneError.disabled } + let surfaceID = IrxSurfaceEventLaneProtocol().normalizedSurfaceID(rawSurfaceID) + let lane = try await openedLane(surfaceID: surfaceID, generation: generation) + useTick &+= 1 + lanes[surfaceID]?.lastUse = useTick + let writer = lane.writer + let result: IrxDeadlineResult + do { + result = try await withIrxDeadlineResult(configuration.stallDeadline) { + try await writer.write(data) + return true + } + } catch { + retire(surfaceID: surfaceID, token: lane.token, errorCode: Self.writeFailedResetCode) + throw error + } + if case .timeout = result { + retire(surfaceID: surfaceID, token: lane.token, errorCode: Self.stalledResetCode) + journal?.record("host-surface-lanes", "write-stalled", ["surface": surfaceID]) + throw LaneError.writeStalled + } + } + + /// Marks the surface the user is interacting with; its lane is scheduled + /// ahead of every other surface and the bulk events lane. + public func noteFocused(surfaceID rawSurfaceID: String) { + let surfaceID = IrxSurfaceEventLaneProtocol().normalizedSurfaceID(rawSurfaceID) + guard !surfaceID.isEmpty, focusedSurfaceID != surfaceID else { return } + let previous = focusedSurfaceID + focusedSurfaceID = surfaceID + if let previous { applyPriority(surfaceID: previous) } + applyPriority(surfaceID: surfaceID) + } + + /// Disabling finishes every lane and refuses new ones until re-enabled. + public func setEnabled(_ enabled: Bool) { + guard isEnabled != enabled else { return } + isEnabled = enabled + if !enabled { finishAll() } + } + + /// Finishes one surface's lane, if the lane is still `generation`'s. + public func close(surfaceID rawSurfaceID: String, generation: UInt64? = nil) { + let surfaceID = IrxSurfaceEventLaneProtocol().normalizedSurfaceID(rawSurfaceID) + guard let lane = lanes[surfaceID], + generation == nil || lane.generation == generation else { return } + lanes.removeValue(forKey: surfaceID) + let writer = lane.writer + Task { await writer.finish() } + } + + /// Finishes every lane and permanently refuses new ones. + public func closeAll() { + isEnabled = false + finishAll() + } + + public func openSurfaceIDs() -> Set { Set(lanes.keys) } + + public func priority(surfaceID rawSurfaceID: String) -> Int32? { + lanes[IrxSurfaceEventLaneProtocol().normalizedSurfaceID(rawSurfaceID)]?.priority + } + + private func openedLane(surfaceID: String, generation: UInt64) async throws -> Lane { + if let lane = lanes[surfaceID] { + if lane.generation == generation { return lane } + // A newer generation means frames on the old stream may be lost; + // never mix the chain across the two streams. + lanes.removeValue(forKey: surfaceID) + let writer = lane.writer + Task { await writer.finish() } + } + evictLeastRecentlyUsedLaneIfFull() + let descriptor = IrxSurfaceEventLaneProtocol().descriptor(surfaceID: surfaceID) + let opener = open + // Opening waits for stream credit. The native open ignores task + // cancellation, so a late stream is released instead of leaked. + let openTask = Task { try await opener(descriptor) } + let result = try await withIrxDeadlineResult(configuration.openDeadline) { + try await openTask.value + } + guard case .operation(let opened?) = result else { + Task { + if let late = try? await openTask.value { + await late.reset(errorCode: Self.supersededResetCode) + } + } + journal?.record("host-surface-lanes", "open-timed-out", ["surface": surfaceID]) + throw LaneError.openTimedOut + } + guard isEnabled else { + await opened.reset(errorCode: Self.supersededResetCode) + throw LaneError.disabled + } + if let raced = lanes[surfaceID], raced.generation == generation { + // A concurrent send for the same generation won the open. + await opened.reset(errorCode: Self.supersededResetCode) + return raced + } + nextToken &+= 1 + let priority = priorityFor(surfaceID: surfaceID) + try? await opened.setPriority(priority) + let lane = Lane( + token: nextToken, + generation: generation, + writer: opened, + priority: priority, + lastUse: useTick + ) + if let replaced = lanes[surfaceID] { + let writer = replaced.writer + Task { await writer.finish() } + } + lanes[surfaceID] = lane + journal?.record( + "host-surface-lanes", "opened", + ["surface": surfaceID, "priority": String(priority), "open": String(lanes.count)] + ) + return lane + } + + private func evictLeastRecentlyUsedLaneIfFull() { + while lanes.count >= configuration.maximumLaneCount, + let victim = lanes.min(by: { $0.value.lastUse < $1.value.lastUse }) { + lanes.removeValue(forKey: victim.key) + let writer = victim.value.writer + Task { await writer.finish() } + journal?.record("host-surface-lanes", "evicted", ["surface": victim.key]) + } + } + + private func retire(surfaceID: String, token: UInt64, errorCode: UInt64) { + guard let lane = lanes[surfaceID], lane.token == token else { return } + lanes.removeValue(forKey: surfaceID) + let writer = lane.writer + // Reset crosses a cancellation-insensitive native bridge; never make + // the caller's recovery wait on it. + Task { await writer.reset(errorCode: errorCode) } + } + + private func finishAll() { + let writers = lanes.values.map(\.writer) + lanes.removeAll() + for writer in writers { + Task { await writer.finish() } + } + } + + private func priorityFor(surfaceID: String) -> Int32 { + surfaceID == focusedSurfaceID + ? configuration.focusedPriority + : configuration.backgroundPriority + } + + private func applyPriority(surfaceID: String) { + guard var lane = lanes[surfaceID] else { return } + let priority = priorityFor(surfaceID: surfaceID) + guard lane.priority != priority else { return } + lane.priority = priority + lanes[surfaceID] = lane + // The native call waits for the lane's in-flight write; apply it + // off the caller's path. Updates for one lane are serialized. + let writer = lane.writer + let previous = lane.priorityUpdate + lanes[surfaceID]?.priorityUpdate = Task { + await previous?.value + try? await writer.setPriority(priority) + } + } +} diff --git a/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxSurfaceEventLaneTests.swift b/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxSurfaceEventLaneTests.swift new file mode 100644 index 000000000000..3f2c761eb471 --- /dev/null +++ b/Packages/Shared/CmuxIrxTransport/Tests/CmuxIrxTransportTests/IrxSurfaceEventLaneTests.swift @@ -0,0 +1,507 @@ +import Foundation +import Testing +@testable import CmuxIrxTransport + +// MARK: - Fakes + +/// In-memory server->client lane read half. Chunks are pushed by the test; +/// `readRaw` waits for the next one, like a QUIC stream with no data yet. +private actor FakeEventLaneReader: IrxEventLaneReading { + private var chunks: [Data] = [] + private var waiter: CheckedContinuation? + private var ended = false + private(set) var stopCodes: [UInt64] = [] + + func push(_ chunk: Data) { + if let waiter { + self.waiter = nil + waiter.resume(returning: chunk) + } else { + chunks.append(chunk) + } + } + + func end() { + ended = true + if let waiter { + self.waiter = nil + waiter.resume(returning: nil) + } + } + + func readRaw() async throws -> Data? { + if !chunks.isEmpty { return chunks.removeFirst() } + if ended { return nil } + return try await withCheckedThrowingContinuation { waiter = $0 } + } + + func stop(errorCode: UInt64) { + stopCodes.append(errorCode) + end() + } +} + +/// Feeds accepted lanes to a hub in the order the test opens them. +private final class FakeLaneAcceptor: @unchecked Sendable { + private let stream: AsyncStream<(IrxLaneDescriptor, any IrxEventLaneReading)> + private let continuation: AsyncStream<(IrxLaneDescriptor, any IrxEventLaneReading)>.Continuation + + init() { + (stream, continuation) = AsyncStream.makeStream() + } + + func open(_ descriptor: IrxLaneDescriptor) -> FakeEventLaneReader { + let reader = FakeEventLaneReader() + continuation.yield((descriptor, reader)) + return reader + } + + func closeConnection() { continuation.finish() } + + var accept: IrxServerEventLaneHub.AcceptLane { + let stream = stream + return { + var iterator = stream.makeAsyncIterator() + return await iterator.next() + } + } +} + +/// In-memory lane write half. A blocked lane accepts no bytes until the test +/// releases it, like a QUIC stream out of flow credit. Like iroh-ffi, every +/// other call on the stream (priority, finish, reset) waits for an in-flight +/// write to finish, because the native stream sits behind one lock. +private actor FakeEventLaneWriter: IrxEventLaneWriting { + let descriptor: IrxLaneDescriptor + private let blocked: Bool + private var blockedWrite: CheckedContinuation? + private var lockWaiters: [CheckedContinuation] = [] + private(set) var written: [Data] = [] + private(set) var priorities: [Int32] = [] + private(set) var finished = false + private(set) var resetCodes: [UInt64] = [] + + init(descriptor: IrxLaneDescriptor, blocked: Bool) { + self.descriptor = descriptor + self.blocked = blocked + } + + func write(_ data: Data) async throws { + await waitForStreamLock() + guard !finished, resetCodes.isEmpty else { throw IrxFrameCodecError.unexpectedEOF } + if blocked { + defer { releaseStreamLock() } + try await withCheckedThrowingContinuation { blockedWrite = $0 } + } + written.append(data) + } + + func setPriority(_ priority: Int32) async { + await waitForStreamLock() + priorities.append(priority) + } + + func finish() async { + await waitForStreamLock() + finished = true + } + + func reset(errorCode: UInt64) async { + await waitForStreamLock() + resetCodes.append(errorCode) + } + + var isWriteBlocked: Bool { blockedWrite != nil } + + /// Fails the stuck write (the connection closing), releasing the lock. + func failBlockedWrite() { + blockedWrite?.resume(throwing: IrxFrameCodecError.unexpectedEOF) + blockedWrite = nil + } + + private func waitForStreamLock() async { + guard blockedWrite != nil else { return } + await withCheckedContinuation { lockWaiters.append($0) } + } + + private func releaseStreamLock() { + let waiters = lockWaiters + lockWaiters.removeAll() + for waiter in waiters { waiter.resume() } + } +} + +private actor FakeLaneOpener { + private(set) var opened: [FakeEventLaneWriter] = [] + var blockedSurfaces: Set = [] + + func block(_ surfaceID: String) { blockedSurfaces.insert(surfaceID) } + func unblock(_ surfaceID: String) { blockedSurfaces.remove(surfaceID) } + + func open(_ descriptor: IrxLaneDescriptor) -> any IrxEventLaneWriting { + let surfaceID = IrxSurfaceEventLaneProtocol().surfaceID(of: descriptor) ?? "" + let writer = FakeEventLaneWriter( + descriptor: descriptor, + blocked: blockedSurfaces.contains(surfaceID) + ) + opened.append(writer) + return writer + } + + func writers(surfaceID: String) -> [FakeEventLaneWriter] { + opened.filter { + IrxSurfaceEventLaneProtocol().surfaceID(of: $0.descriptor) == surfaceID + } + } +} + +private func frame(_ text: String) -> Data { + var length = UInt32(text.utf8.count).bigEndian + var data = Data(bytes: &length, count: 4) + data.append(Data(text.utf8)) + return data +} + +private func decodeFrames(_ data: Data) -> [String] { + var buffer = data + var result: [String] = [] + while buffer.count >= 4 { + let length = buffer.prefix(4).reduce(0) { ($0 << 8) | Int($1) } + guard buffer.count >= 4 + length else { break } + result.append(String(decoding: buffer.dropFirst(4).prefix(length), as: UTF8.self)) + buffer.removeFirst(4 + length) + } + return result +} + +private func waitUntil( + _ condition: @escaping @Sendable () async -> Bool +) async throws -> Bool { + let reached = try await withIrxDeadline(.seconds(2), onTimeout: {}) { + while !Task.isCancelled { + if await condition() { return true } + try await Task.sleep(for: .milliseconds(2)) + } + return false + } + return reached == true +} + +private actor FrameCollector { + private(set) var frames: [String] = [] + func append(_ data: Data) { frames.append(contentsOf: decodeFrames(data)) } +} + +private func collect(_ stream: IrxServerEventLaneHub.Output, into collector: FrameCollector) -> Task { + Task { + do { + for try await chunk in stream { await collector.append(chunk) } + } catch {} + } +} + +// MARK: - Client hub + +@Suite(.timeLimit(.minutes(1))) +struct IrxServerEventLaneHubTests { + @Test func surfaceFrameIsDeliveredWhileAnotherLaneIsStalledMidFrame() async throws { + let acceptor = FakeLaneAcceptor() + let hub = IrxServerEventLaneHub(acceptLane: acceptor.accept) + let collector = FrameCollector() + let consumer = collect(await hub.subscribe(), into: collector) + defer { consumer.cancel() } + + let shared = acceptor.open(IrxLaneDescriptor(lane: .events)) + let busy = acceptor.open(IrxSurfaceEventLaneProtocol().descriptor(surfaceID: "A")) + let typed = acceptor.open(IrxSurfaceEventLaneProtocol().descriptor(surfaceID: "B")) + + // Surface A's large replay has only partly arrived; its lane is + // waiting for the rest. Surface B's echo must not wait behind it. + let replay = frame(String(repeating: "a", count: 64 * 1024)) + await busy.push(replay.prefix(10_000)) + await shared.push(frame("workspace.updated")) + await typed.push(frame("echo-b")) + + #expect(try await waitUntil { await collector.frames.contains("echo-b") }) + #expect(await collector.frames.sorted() == ["echo-b", "workspace.updated"]) + + await busy.push(replay.dropFirst(10_000)) + #expect(try await waitUntil { await collector.frames.count == 3 }) + await hub.stop() + } + + @Test func framesSplitAcrossChunksArriveWholeAndInLaneOrder() async throws { + let acceptor = FakeLaneAcceptor() + let hub = IrxServerEventLaneHub(acceptLane: acceptor.accept) + let collector = FrameCollector() + let consumer = collect(await hub.subscribe(), into: collector) + defer { consumer.cancel() } + + let shared = acceptor.open(IrxLaneDescriptor(lane: .events)) + let surface = acceptor.open(IrxSurfaceEventLaneProtocol().descriptor(surfaceID: "S")) + var surfaceBytes = Data() + for index in 0..<20 { surfaceBytes.append(frame("s\(index)")) } + var sharedBytes = Data() + for index in 0..<20 { sharedBytes.append(frame("e\(index)")) } + // Interleave odd-sized chunks from both lanes. + var surfaceOffset = 0 + var sharedOffset = 0 + while surfaceOffset < surfaceBytes.count || sharedOffset < sharedBytes.count { + if surfaceOffset < surfaceBytes.count { + let end = min(surfaceBytes.count, surfaceOffset + 7) + await surface.push(surfaceBytes.subdata(in: surfaceOffset.. IrxSurfaceEventLanes { + IrxSurfaceEventLanes(configuration: configuration) { descriptor in + await opener.open(descriptor) + } + } + + @Test func stalledSurfaceDoesNotDelayAnotherSurfacesWrite() async throws { + let opener = FakeLaneOpener() + await opener.block("a") + let lanes = makeLanes(opener) + let stalled = Task { try await lanes.send(frame("replay-a"), surfaceID: "A", generation: 0) } + #expect(try await waitUntil { + guard let writer = await opener.writers(surfaceID: "a").first else { return false } + return await writer.isWriteBlocked + }) + try await lanes.send(frame("echo-b"), surfaceID: "B", generation: 0) + let bWriter = try #require(await opener.writers(surfaceID: "b").first) + #expect(await bWriter.written == [frame("echo-b")]) + await lanes.closeAll() + await opener.writers(surfaceID: "a").first?.failBlockedWrite() + _ = await stalled.result + } + + @Test func stalledWriteResetsTheStreamAndTheNextSendReopens() async throws { + let opener = FakeLaneOpener() + await opener.block("a") + let lanes = makeLanes(opener, configuration: .init(stallDeadline: .milliseconds(50))) + await #expect(throws: IrxSurfaceEventLanes.LaneError.writeStalled) { + try await lanes.send(frame("x"), surfaceID: "a", generation: 0) + } + let stalledWriter = try #require(await opener.writers(surfaceID: "a").first) + #expect(await lanes.openSurfaceIDs().isEmpty) + + // The stuck stream's reset waits behind its write, so recovery must + // not: the next frame goes out on a fresh stream right away. + await opener.unblock("a") + try await lanes.send(frame("full"), surfaceID: "a", generation: 1) + let writers = await opener.writers(surfaceID: "a") + #expect(writers.count == 2) + #expect(await writers[1].written == [frame("full")]) + #expect(await stalledWriter.resetCodes.isEmpty) + + // Once the stuck write fails, the queued reset lands on the old stream. + await stalledWriter.failBlockedWrite() + #expect(try await waitUntil { + await stalledWriter.resetCodes == [IrxSurfaceEventLanes.stalledResetCode] + }) + } + + @Test func newGenerationFinishesTheOldStreamAndOpensAFreshOne() async throws { + let opener = FakeLaneOpener() + let lanes = makeLanes(opener) + try await lanes.send(frame("1"), surfaceID: "s", generation: 0) + try await lanes.send(frame("2"), surfaceID: "s", generation: 0) + try await lanes.send(frame("3"), surfaceID: "s", generation: 1) + let writers = await opener.writers(surfaceID: "s") + #expect(writers.count == 2) + #expect(await writers[0].written == [frame("1"), frame("2")]) + #expect(try await waitUntil { await writers[0].finished }) + #expect(await writers[1].written == [frame("3")]) + } + + @Test func focusedSurfaceIsScheduledAboveEveryOtherLane() async throws { + let opener = FakeLaneOpener() + let lanes = makeLanes(opener) + await lanes.noteFocused(surfaceID: "A") + try await lanes.send(frame("a"), surfaceID: "a", generation: 0) + try await lanes.send(frame("b"), surfaceID: "b", generation: 0) + #expect(await lanes.priority(surfaceID: "a") == 100) + #expect(await lanes.priority(surfaceID: "b") == 50) + + await lanes.noteFocused(surfaceID: "b") + #expect(await lanes.priority(surfaceID: "a") == 50) + #expect(await lanes.priority(surfaceID: "b") == 100) + let bWriter = try #require(await opener.writers(surfaceID: "b").first) + #expect(try await waitUntil { await bWriter.priorities == [50, 100] }) + } + + @Test func notingFocusNeverWaitsForAStalledWrite() async throws { + let opener = FakeLaneOpener() + await opener.block("a") + let lanes = makeLanes(opener) + let stalled = Task { try await lanes.send(frame("replay-a"), surfaceID: "a", generation: 0) } + #expect(try await waitUntil { + guard let writer = await opener.writers(surfaceID: "a").first else { return false } + return await writer.isWriteBlocked + }) + // Input on surface A marks it focused; that must return at once even + // though A's lane is mid-write. + await lanes.noteFocused(surfaceID: "a") + #expect(await lanes.priority(surfaceID: "a") == 100) + let aWriter = try #require(await opener.writers(surfaceID: "a").first) + #expect(await aWriter.priorities == [50]) + await aWriter.failBlockedWrite() + #expect(try await waitUntil { await aWriter.priorities == [50, 100] }) + _ = await stalled.result + await lanes.closeAll() + } + + @Test func laneCountIsBoundedByEvictingTheLeastRecentlyUsedLane() async throws { + let opener = FakeLaneOpener() + let lanes = makeLanes(opener, configuration: .init(maximumLaneCount: 2)) + try await lanes.send(frame("1"), surfaceID: "one", generation: 0) + try await lanes.send(frame("2"), surfaceID: "two", generation: 0) + try await lanes.send(frame("1b"), surfaceID: "one", generation: 0) + try await lanes.send(frame("3"), surfaceID: "three", generation: 0) + #expect(await lanes.openSurfaceIDs() == ["one", "three"]) + let evicted = try #require(await opener.writers(surfaceID: "two").first) + #expect(try await waitUntil { await evicted.finished }) + } + + @Test func disabledLanesRefuseToOpen() async throws { + let opener = FakeLaneOpener() + let lanes = makeLanes(opener) + try await lanes.send(frame("1"), surfaceID: "s", generation: 0) + await lanes.setEnabled(false) + await #expect(throws: IrxSurfaceEventLanes.LaneError.disabled) { + try await lanes.send(frame("2"), surfaceID: "s", generation: 0) + } + #expect(await lanes.openSurfaceIDs().isEmpty) + let writer = try #require(await opener.writers(surfaceID: "s").first) + #expect(try await waitUntil { await writer.finished }) + } + + @Test func surfaceLaneDescriptorRoundTripsAndSharedLaneHasNoSurface() { + let descriptor = IrxSurfaceEventLaneProtocol().descriptor(surfaceID: " ABC-def ") + #expect(descriptor.lane == .events) + #expect(IrxSurfaceEventLaneProtocol().surfaceID(of: descriptor) == "abc-def") + #expect(IrxSurfaceEventLaneProtocol().surfaceID(of: IrxLaneDescriptor(lane: .events)) == nil) + #expect(IrxSurfaceEventLaneProtocol().surfaceID( + of: IrxLaneDescriptor(lane: .terminal, resource: "terminal:x") + ) == nil) + } + + @Test func frameAlignerReturnsOnlyCompleteFrames() throws { + var aligner = IrxEventFrameAligner(maximumFrameByteCount: 1024) + let bytes = frame("hello") + frame("world") + #expect(try aligner.append(bytes.prefix(3)) == nil) + let first = try #require(try aligner.append(bytes.subdata(in: 3..<12))) + #expect(decodeFrames(first) == ["hello"]) + #expect(aligner.hasPartialFrame) + let second = try #require(try aligner.append(bytes.dropFirst(12))) + #expect(decodeFrames(second) == ["world"]) + #expect(!aligner.hasPartialFrame) + + var small = IrxEventFrameAligner(maximumFrameByteCount: 2) + #expect(throws: IrxEventFrameAligner.Failure.frameTooLarge(5)) { + _ = try small.append(frame("hello")) + } + } +} diff --git a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCClient.swift b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCClient.swift index d5e655b43b6d..c5adf1e89968 100644 --- a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCClient.swift +++ b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileCoreRPCClient.swift @@ -430,6 +430,11 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable { } } + /// Wire opt-in for one server->client stream per terminal surface (see + /// `IrxSurfaceEventLaneProtocol` on the host). + static let surfaceEventLanesParameterKey = "surface_event_lanes" + static let surfaceEventLanesParameterValue = "v1" + /// Adds the rolling-compatible opt-in only after the Iroh accept owner is /// installed. Older hosts ignore the field and continue control delivery. private func requestAdvertisingIndependentEvents( @@ -455,6 +460,11 @@ public final class MobileCoreRPCClient: MobileSyncing, Sendable { return requestData } params["event_transport"] = "iroh_server_events_v1" + if runtime.independentEventsMergeSurfaceLanes { + // Older hosts ignore the field and keep render-grid output on the + // shared events lane; newer hosts echo it when they granted lanes. + params[Self.surfaceEventLanesParameterKey] = Self.surfaceEventLanesParameterValue + } request["params"] = params return (try? JSONSerialization.data(withJSONObject: request)) ?? requestData } diff --git a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileEventSubscribeResponse.swift b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileEventSubscribeResponse.swift index 1abfe9216544..d6666203e0a9 100644 --- a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileEventSubscribeResponse.swift +++ b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileEventSubscribeResponse.swift @@ -25,10 +25,15 @@ public struct MobileEventSubscribeResponse: Decodable, Sendable { /// The selected event delivery path, absent on older hosts. public let eventTransport: String? + /// `"v1"` when the host put each terminal's render-grid output on its own + /// stream; absent on older hosts and whenever the host kept one lane. + public let surfaceEventLanes: String? + private enum CodingKeys: String, CodingKey { case streamID = "stream_id" case alreadySubscribed = "already_subscribed" case eventTransport = "event_transport" + case surfaceEventLanes = "surface_event_lanes" } /// Decode a subscribe acknowledgement from raw JSON data. @@ -48,5 +53,6 @@ public struct MobileEventSubscribeResponse: Decodable, Sendable { streamID = (try container.decodeIfPresent(String.self, forKey: .streamID)) ?? "" alreadySubscribed = try container.decodeIfPresent(Bool.self, forKey: .alreadySubscribed) eventTransport = try container.decodeIfPresent(String.self, forKey: .eventTransport) + surfaceEventLanes = try container.decodeIfPresent(String.self, forKey: .surfaceEventLanes) } } diff --git a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileSyncRuntime.swift b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileSyncRuntime.swift index 7448614c5c4a..6228b51ce079 100644 --- a/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileSyncRuntime.swift +++ b/Packages/iOS/CmuxMobileRPC/Sources/CmuxMobileRPC/MobileSyncRuntime.swift @@ -39,6 +39,12 @@ public protocol MobileSyncRuntime: Sendable { /// Optional Iroh-only source for independently framed server events. /// A nil provider preserves control-stream delivery for every route. var independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? { get } + /// Whether ``independentEventByteStreamProvider`` reads every server + /// event lane the host opens (one per terminal surface) and forwards them + /// frame-aligned. Only then may the client ask the host to put each + /// terminal's render-grid output on its own stream; a single-lane reader + /// would never see a second stream. + var independentEventsMergeSurfaceLanes: Bool { get } /// Optional source for one independent, sequence-aware terminal lane per /// mounted surface. A nil provider preserves control/event delivery. var terminalLaneProvider: MobileTerminalLaneProvider? { get } @@ -71,6 +77,7 @@ public protocol MobileSyncRuntime: Sendable { public extension MobileSyncRuntime { var independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? { nil } + var independentEventsMergeSurfaceLanes: Bool { false } var terminalLaneProvider: MobileTerminalLaneProvider? { nil } var terminalInputLaneProvider: MobileTerminalLaneProvider? { nil } var artifactLaneProvider: MobileArtifactLaneProvider? { nil } diff --git a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileCoreRPCIndependentEventTests.swift b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileCoreRPCIndependentEventTests.swift index 45c7524a8da9..437132a7a0c5 100644 --- a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileCoreRPCIndependentEventTests.swift +++ b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileCoreRPCIndependentEventTests.swift @@ -90,6 +90,40 @@ struct MobileCoreRPCIndependentEventTests { await client.disconnect() } + @Test(arguments: [false, true]) + func subscribeRequestsSurfaceEventLanesOnlyWhenTheReaderMergesThem( + merges: Bool + ) async throws { + let route = try irohRoute(hexBytePair: merges ? "9c" : "9b") + let source = IndependentEventSource() + let transport = SubscribeRoundTripTransport() + let runtime = TestMobileSyncRuntime( + transportFactory: FixedTransportFactory(transport: transport), + independentEventByteStreamProvider: { _ in await source.makeStream() }, + independentEventsMergeSurfaceLanes: merges + ) + let client = MobileCoreRPCClient( + runtime: runtime, + route: route, + ticket: try ticket(route: route, deviceSuffix: merges ? "013" : "012") + ) + let request = try MobileCoreRPCClient.requestData( + method: "mobile.events.subscribe", + params: [ + "stream_id": "events", + "topics": ["terminal.render_grid"], + ] + ) + + _ = try await client.sendRequest(request) + + // A reader that sees only one stream would never read a per-surface + // stream, so the opt-in rides only with a merging reader. + #expect(await transport.recordedEventTransport() == "iroh_server_events_v1") + #expect(await transport.recordedSurfaceEventLaneRequest() == (merges ? "v1" : nil)) + await client.disconnect() + } + @Test func independentlyFramedEventsReachTheExistingListenerPipeline() async throws { let route = try irohRoute(hexBytePair: "ab") @@ -458,6 +492,7 @@ private actor SubscribeRoundTripTransport: CmxByteTransport, CmxByteTransportClo private var waiter: CheckedContinuation? private var closeWaiters: [CheckedContinuation] = [] private var eventTransports: [String?] = [] + private var surfaceEventLaneRequests: [String?] = [] private var closed = false func connect() async throws {} @@ -479,6 +514,7 @@ private actor SubscribeRoundTripTransport: CmxByteTransport, CmxByteTransportClo let params = request["params"] as? [String: Any] let eventTransport = params?["event_transport"] as? String eventTransports.append(eventTransport) + surfaceEventLaneRequests.append(params?["surface_event_lanes"] as? String) let response = try JSONSerialization.data(withJSONObject: [ "id": request["id"] ?? NSNull(), "ok": true, @@ -528,4 +564,8 @@ private actor SubscribeRoundTripTransport: CmxByteTransport, CmxByteTransportClo } func recordedEventTransports() -> [String?] { eventTransports } + + func recordedSurfaceEventLaneRequest() -> String? { + surfaceEventLaneRequests.last ?? nil + } } diff --git a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileTerminalDTODecodeTests.swift b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileTerminalDTODecodeTests.swift index 6ca23f5b3404..fed68807c69f 100644 --- a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileTerminalDTODecodeTests.swift +++ b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/MobileTerminalDTODecodeTests.swift @@ -27,6 +27,18 @@ import Testing #expect(response.eventTransport == "iroh_server_events_v1") } + @Test func subscribeResponseDecodesSurfaceEventLaneGrant() throws { + let granted = try MobileEventSubscribeResponse.decode(Data( + #"{"stream_id":"s","event_transport":"iroh_server_events_v1","surface_event_lanes":"v1"}"#.utf8 + )) + #expect(granted.surfaceEventLanes == "v1") + // Older hosts omit the field: render-grid stays on the shared lane. + let older = try MobileEventSubscribeResponse.decode(Data( + #"{"stream_id":"s","event_transport":"iroh_server_events_v1"}"#.utf8 + )) + #expect(older.surfaceEventLanes == nil) + } + @Test func hostStatusDecodesRenderGridCapability() throws { let data = Data(#"{"capabilities":["terminal.render_grid.v1"],"terminal_fidelity":"render_grid"}"#.utf8) let response = try MobileHostStatusResponse.decode(data) diff --git a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/TransportTestDoubles.swift b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/TransportTestDoubles.swift index 9eddb418dfdf..696d6231f3de 100644 --- a/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/TransportTestDoubles.swift +++ b/Packages/iOS/CmuxMobileRPC/Tests/CmuxMobileRPCTests/TransportTestDoubles.swift @@ -16,6 +16,7 @@ struct TestMobileSyncRuntime: MobileSyncRuntime { var now: @Sendable () -> Date var supportsServerPushEvents: Bool var independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? + var independentEventsMergeSurfaceLanes: Bool init( transportFactory: any CmxByteTransportFactory, @@ -28,8 +29,10 @@ struct TestMobileSyncRuntime: MobileSyncRuntime { pairingRequestTimeoutNanoseconds: UInt64 = 30 * 1_000_000_000, now: @escaping @Sendable () -> Date = Date.init, supportsServerPushEvents: Bool = true, - independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? = nil + independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? = nil, + independentEventsMergeSurfaceLanes: Bool = false ) { + self.independentEventsMergeSurfaceLanes = independentEventsMergeSurfaceLanes self.supportedRouteKinds = supportedRouteKinds self.transportFactory = transportFactory self.stackAccessTokenProvider = stackAccessTokenProvider ?? { diff --git a/Sources/Mobile/MobileHostConnectionEventQueue.swift b/Sources/Mobile/MobileHostConnectionEventQueue.swift index 9a5d7b2b8435..75a90759c838 100644 --- a/Sources/Mobile/MobileHostConnectionEventQueue.swift +++ b/Sources/Mobile/MobileHostConnectionEventQueue.swift @@ -49,12 +49,25 @@ enum MobileHostEventTopicPolicy { } } +/// Delivery lane of one queued event. Every lane has its own drain, so a +/// stalled write on one lane never delays events queued on another. +/// +/// `.shared` is the ordered events path (independent events stream or the +/// control stream). `.surface` carries one terminal's render-grid frames on +/// its own QUIC stream once the client negotiated surface event lanes. +enum MobileHostEventLane: Hashable, Sendable { + case shared + case surface(String) +} + /// Outcome of one synchronous admission attempt on a connection's event queue. struct MobileHostEventEnqueueResult: Sendable { /// The event was appended to the queue. let admitted: Bool - /// The caller must start the (single) drain task for this connection. + /// The caller must start the drain task for ``drainLane``. let startDrain: Bool + /// The lane whose drain the caller must start when ``startDrain`` is set. + var drainLane: MobileHostEventLane = .shared /// Surfaces whose queued render-grid frames were shed; the caller must ask /// the producer for a full-frame resync of each. let renderGridResyncSurfaceIDs: Set @@ -103,8 +116,24 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { let coalesceKey: String? let frame: Data let stateSeq: UInt64? + var lane: MobileHostEventLane = .shared + /// Stream generation of a surface lane. A new generation means a new + /// QUIC stream, so the render-grid chain must re-base on it. + var laneGeneration: UInt64 = 0 + } + + /// The stream a surface's render-grid chain was last admitted on. Frames + /// on two different streams can arrive in either order, so a delta may + /// only follow a frame that travelled the same route. + private enum RenderGridRoute: Equatable { + case shared + case surface(generation: UInt64) } + /// Consecutive failures after which a surface stops using its own lane + /// and rides the shared lane until surface lanes are renegotiated. + static let maximumSurfaceLaneFailureCount = 3 + static let defaultMaximumEventCount = 256 static let defaultMaximumByteCount = MobileSyncFrameCodec.defaultMaximumFrameByteCount @@ -116,8 +145,19 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { private var subscribedTopics: Set = [] private var queuedEvents: [QueuedEvent] = [] private var queuedByteCount = 0 - private var drainActive = false + /// Lanes with a running drain. At most one drain per lane. + private var drainingLanes: Set = [] private var isClosed = false + /// Maximum concurrently assigned surface lanes; 0 disables surface lanes. + private var surfaceLaneLimit = 0 + /// Assigned surface lanes and their last-use tick (for LRU reassignment). + private var surfaceLaneLastUse: [String: UInt64] = [:] + private var surfaceLaneUseTick: UInt64 = 0 + private var surfaceLaneGenerations: [String: UInt64] = [:] + private var surfaceLaneFailureCounts: [String: Int] = [:] + private var sharedLanePinnedSurfaceIDs: Set = [] + private var queuedCountByLane: [MobileHostEventLane: Int] = [:] + private var lastRenderGridRouteBySurfaceID: [String: RenderGridRoute] = [:] /// Surfaces whose delta chain was broken by a shed frame. Only a /// full-frame render-grid event readmits the surface; deltas are refused so /// the client can never apply a delta whose predecessor was dropped. @@ -188,7 +228,30 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { lock.unlock() return .rejected } + let (lane, laneGeneration) = laneLocked(topic: topic, coalesceKey: coalesceKey) var resyncSurfaceIDs = Set() + if isRenderGrid, let coalesceKey, !isFullRenderGridFrame { + let route: RenderGridRoute = lane == .shared + ? .shared + : .surface(generation: laneGeneration) + if let previousRoute = lastRenderGridRouteBySurfaceID[coalesceKey], + previousRoute != route { + // This delta builds on a frame that travelled another stream, + // which may still be in flight behind it. Re-base the chain + // with a full frame on the new route instead. + poisonedRenderGridSurfaceIDs.insert(coalesceKey) + lock.unlock() + return MobileHostEventEnqueueResult( + admitted: false, + startDrain: false, + renderGridResyncSurfaceIDs: [coalesceKey], + depthAfterEnqueue: nil, + shedEventCount: 0, + shedByteCount: 0, + simulatorFrameShedPanelIDs: [] + ) + } + } var shedSummary = MobileHostEventShedSummary() if !hasRoomLocked(for: frame) { shedSummary = shedDroppableEventsLocked(for: frame, resyncSurfaceIDs: &resyncSurfaceIDs) @@ -238,23 +301,29 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { topic: topic, coalesceKey: coalesceKey, frame: frame, - stateSeq: stateSeq + stateSeq: stateSeq, + lane: lane, + laneGeneration: laneGeneration ) ) queuedByteCount += frame.count + queuedCountByLane[lane, default: 0] += 1 let depthAfterEnqueue = queuedEvents.count + if isRenderGrid, let coalesceKey { + lastRenderGridRouteBySurfaceID[coalesceKey] = lane == .shared + ? .shared + : .surface(generation: laneGeneration) + } if isRenderGrid, isFullRenderGridFrame, let coalesceKey { poisonedRenderGridSurfaceIDs.remove(coalesceKey) resyncAfterDrainSurfaceIDs.remove(coalesceKey) } - let startDrain = !drainActive - if startDrain { - drainActive = true - } + let startDrain = drainingLanes.insert(lane).inserted lock.unlock() return MobileHostEventEnqueueResult( admitted: true, startDrain: startDrain, + drainLane: lane, renderGridResyncSurfaceIDs: resyncSurfaceIDs, depthAfterEnqueue: depthAfterEnqueue, shedEventCount: shedSummary.eventCount, @@ -263,44 +332,193 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { ) } - func dequeue() -> QueuedEvent? { + /// Removes the oldest event queued on `lane`. Events on other lanes keep + /// their global order for shedding. + func dequeue(lane: MobileHostEventLane = .shared) -> QueuedEvent? { lock.lock() defer { lock.unlock() } - guard !queuedEvents.isEmpty else { return nil } - let event = queuedEvents.removeFirst() + guard queuedCountByLane[lane, default: 0] > 0, + let index = queuedEvents.firstIndex(where: { $0.lane == lane }) else { + return nil + } + let event = queuedEvents.remove(at: index) queuedByteCount -= event.frame.count + decrementLaneCountLocked(lane) return event } - /// Called by the drain loop after `dequeue` returned nil. Returns true when - /// events raced in and the loop must keep draining; otherwise the drain is - /// marked finished so the next enqueue can claim a fresh one. - func finishDrain() -> Bool { + /// Called by a lane's drain loop after `dequeue` returned nil. Returns + /// true when events raced in and the loop must keep draining; otherwise + /// the drain is marked finished so the next enqueue can claim a fresh one. + func finishDrain(lane: MobileHostEventLane = .shared) -> Bool { lock.lock() defer { lock.unlock() } - if queuedEvents.isEmpty || isClosed { - drainActive = false + if queuedCountByLane[lane, default: 0] == 0 || isClosed { + drainingLanes.remove(lane) return false } return true } - /// Marks the drain inactive after an abnormal exit (close, lane + /// Marks the lane's drain inactive after an abnormal exit (close, lane /// negotiation, failed delivery) so a later enqueue can claim a fresh one. - func abandonDrain() { + func abandonDrain(lane: MobileHostEventLane = .shared) { lock.lock() - drainActive = false + drainingLanes.remove(lane) lock.unlock() } - /// Claims the drain when events are pending and none is running (used when - /// independent-lane negotiation finishes and delivery may resume). + /// Claims the shared lane's drain when events are pending and none is + /// running (used when independent-lane negotiation finishes). func claimDrain() -> Bool { + claimDrains().contains(.shared) + } + + /// Claims every lane with pending events and no running drain. The caller + /// must start one drain per returned lane. + func claimDrains() -> [MobileHostEventLane] { lock.lock() defer { lock.unlock() } - guard !isClosed, !drainActive, !queuedEvents.isEmpty else { return false } - drainActive = true - return true + guard !isClosed else { return [] } + var claimed: [MobileHostEventLane] = [] + for (lane, count) in queuedCountByLane where count > 0 { + if drainingLanes.insert(lane).inserted { + claimed.append(lane) + } + } + return claimed + } + + // MARK: Surface lanes + + /// Routes future render-grid frames onto per-surface lanes, at most + /// `limit` at once. Surfaces beyond the limit ride the shared lane. + func enableSurfaceLanes(limit: Int) { + lock.lock() + surfaceLaneLimit = max(0, limit) + sharedLanePinnedSurfaceIDs.removeAll() + surfaceLaneFailureCounts.removeAll() + lock.unlock() + } + + var surfaceLanesEnabled: Bool { + lock.lock() + defer { lock.unlock() } + return surfaceLaneLimit > 0 + } + + /// Returns every future event to the shared lane. Frames still queued for + /// a surface lane are dropped (that lane is no longer drained) and their + /// surfaces are poisoned; the caller must request a full resync for each + /// returned surface. + func disableSurfaceLanes() -> Set { + lock.lock() + defer { lock.unlock() } + guard surfaceLaneLimit > 0 else { return [] } + surfaceLaneLimit = 0 + surfaceLaneLastUse.removeAll() + var resync = Set() + var droppedByteCount = 0 + queuedEvents.removeAll { event in + guard case .surface(let surfaceID) = event.lane else { return false } + resync.insert(surfaceID) + droppedByteCount += event.frame.count + return true + } + queuedByteCount -= droppedByteCount + for lane in queuedCountByLane.keys where lane != .shared { + queuedCountByLane.removeValue(forKey: lane) + } + poisonedRenderGridSurfaceIDs.formUnion(resync) + return resync + } + + /// Records that a surface lane stream failed or stalled. Frames written + /// to it may be lost, so the surface's queued frames are dropped, its + /// chain is poisoned, and the next stream gets a new generation. Returns + /// the surfaces that need a full-frame resync. A stale `generation` + /// (already retired) changes nothing. + func retireSurfaceLane(surfaceID: String, generation: UInt64) -> Set { + lock.lock() + defer { lock.unlock() } + guard !isClosed, surfaceLaneGenerations[surfaceID, default: 0] == generation else { + return [] + } + surfaceLaneGenerations[surfaceID] = generation &+ 1 + surfaceLaneLastUse.removeValue(forKey: surfaceID) + let failures = surfaceLaneFailureCounts[surfaceID, default: 0] + 1 + surfaceLaneFailureCounts[surfaceID] = failures + if failures >= Self.maximumSurfaceLaneFailureCount { + sharedLanePinnedSurfaceIDs.insert(surfaceID) + } + var droppedByteCount = 0 + queuedEvents.removeAll { event in + guard event.topic == MobileHostEventTopicPolicy.renderGridTopic, + event.coalesceKey == surfaceID else { return false } + droppedByteCount += event.frame.count + decrementLaneCountLocked(event.lane) + return true + } + queuedByteCount -= droppedByteCount + poisonedRenderGridSurfaceIDs.insert(surfaceID) + return [surfaceID] + } + + /// Clears a surface's consecutive-failure count after a delivered frame. + func noteSurfaceLaneDelivered(surfaceID: String) { + lock.lock() + if surfaceLaneFailureCounts[surfaceID] != nil { + surfaceLaneFailureCounts.removeValue(forKey: surfaceID) + } + lock.unlock() + } + + /// Current generation of a surface's lane stream. + func surfaceLaneGeneration(surfaceID: String) -> UInt64 { + lock.lock() + defer { lock.unlock() } + return surfaceLaneGenerations[surfaceID, default: 0] + } + + private func laneLocked( + topic: String, + coalesceKey: String? + ) -> (MobileHostEventLane, UInt64) { + guard surfaceLaneLimit > 0, + topic == MobileHostEventTopicPolicy.renderGridTopic, + let surfaceID = coalesceKey, + !sharedLanePinnedSurfaceIDs.contains(surfaceID) else { + return (.shared, 0) + } + surfaceLaneUseTick &+= 1 + if surfaceLaneLastUse[surfaceID] == nil { + if surfaceLaneLastUse.count >= surfaceLaneLimit { + // Reassign the least recently used idle lane; a lane with + // queued or in-flight frames keeps its surface. + let idle = surfaceLaneLastUse.filter { entry in + let lane = MobileHostEventLane.surface(entry.key) + return queuedCountByLane[lane, default: 0] == 0 + && !drainingLanes.contains(lane) + } + guard let victim = idle.min(by: { $0.value < $1.value })?.key else { + return (.shared, 0) + } + surfaceLaneLastUse.removeValue(forKey: victim) + // The victim's next frame opens a new stream. + surfaceLaneGenerations[victim, default: 0] &+= 1 + } + } + surfaceLaneLastUse[surfaceID] = surfaceLaneUseTick + return (.surface(surfaceID), surfaceLaneGenerations[surfaceID, default: 0]) + } + + private func decrementLaneCountLocked(_ lane: MobileHostEventLane) { + let remaining = queuedCountByLane[lane, default: 0] - 1 + if remaining > 0 { + queuedCountByLane[lane] = remaining + } else { + queuedCountByLane.removeValue(forKey: lane) + } } /// Poisoned surfaces whose full-frame resync should be re-requested now @@ -346,6 +564,10 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { resyncAfterDrainSurfaceIDs.removeAll() simulatorFrameReplayAfterDrainPanelIDs.removeAll() subscribedTopics.removeAll() + queuedCountByLane.removeAll() + surfaceLaneLimit = 0 + surfaceLaneLastUse.removeAll() + lastRenderGridRouteBySurfaceID.removeAll() lock.unlock() } @@ -371,6 +593,7 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { } queuedEvents.remove(at: index) queuedByteCount -= event.frame.count + decrementLaneCountLocked(event.lane) summary.record(event) if event.topic == MobileHostEventTopicPolicy.renderGridTopic, let surfaceID = event.coalesceKey, @@ -393,6 +616,7 @@ final class MobileHostConnectionEventQueue: @unchecked Sendable { } summary.record(event) cascadeByteCount += event.frame.count + decrementLaneCountLocked(event.lane) return true } queuedByteCount -= cascadeByteCount diff --git a/Sources/Mobile/MobileHostIrxEventWriter.swift b/Sources/Mobile/MobileHostIrxEventWriter.swift index 5403fc23b060..ab7489ed4888 100644 --- a/Sources/Mobile/MobileHostIrxEventWriter.swift +++ b/Sources/Mobile/MobileHostIrxEventWriter.swift @@ -3,14 +3,33 @@ import Foundation /// Server-events lane writer over irx: opened lazily at priority 50, reset on /// stall so the host service can renegotiate, mirroring the legacy contract. +/// +/// When the phone negotiated surface lanes, each terminal's render-grid frames +/// go on their own uni stream (``IrxSurfaceEventLanes``) instead of this +/// shared lane, so a burst for one terminal cannot delay another's echo. actor MobileHostIrxEventWriter: MobileHostIndependentEventWriting { private let connection: IrxConnection private let journal: IrxJournal + private let surfaceLanes: IrxSurfaceEventLanes private var writer: IrxStreamWriter? - init(connection: IrxConnection, journal: IrxJournal) { + nonisolated let maximumSurfaceEventLaneCount: Int + + init( + connection: IrxConnection, + journal: IrxJournal, + surfaceLaneConfiguration: IrxSurfaceEventLanes.Configuration = .init() + ) { self.connection = connection self.journal = journal + maximumSurfaceEventLaneCount = surfaceLaneConfiguration.maximumLaneCount + surfaceLanes = IrxSurfaceEventLanes( + configuration: surfaceLaneConfiguration, + journal: journal, + open: { descriptor in + try await connection.openUniLane(descriptor) + } + ) } func probe(_ framedData: Data) async -> Bool { @@ -36,12 +55,26 @@ actor MobileHostIrxEventWriter: MobileHostIndependentEventWriting { } func close() async { + await surfaceLanes.closeAll() if let writer { await writer.finish() } writer = nil } + func sendSurfaceEvent(_ framedData: Data, surfaceID: String, generation: UInt64) async throws { + try await surfaceLanes.send(framedData, surfaceID: surfaceID, generation: generation) + } + + func setSurfaceEventLanesEnabled(_ enabled: Bool) async { + await surfaceLanes.setEnabled(enabled) + journal.record("host-events", enabled ? "surface-lanes-enabled" : "surface-lanes-disabled") + } + + func noteInteractiveSurface(_ surfaceID: String) async { + await surfaceLanes.noteFocused(surfaceID: surfaceID) + } + private func openedWriter() async throws -> IrxStreamWriter { if let writer { return writer } let opened = try await connection.openUniLane(IrxLaneDescriptor(lane: .events)) diff --git a/Sources/Mobile/MobileHostIrxRuntime.swift b/Sources/Mobile/MobileHostIrxRuntime.swift index 19f52370bbcc..ae50771d09b4 100644 --- a/Sources/Mobile/MobileHostIrxRuntime.swift +++ b/Sources/Mobile/MobileHostIrxRuntime.swift @@ -977,7 +977,12 @@ final class MobileHostIrxRuntime: MobileHostPairingRuntime { let laneLoop = Task { await Self.runLaneLoop( irx, admittedPeer: admittedPeer, artifactRegistry: artifactRegistry, - journal: journal) + journal: journal, + onInteractiveSurface: { surfaceID in + // Fire-and-forget: input delivery never waits on the + // output side. Keystrokes arrive at human rate. + Task { await eventWriter.noteInteractiveSurface(surfaceID.uuidString) } + }) } let controlTransport = IrxControlByteTransport( connection: irx, control: control, closeCode: .hostShutdown) @@ -1029,7 +1034,8 @@ final class MobileHostIrxRuntime: MobileHostPairingRuntime { _ irx: IrxConnection, admittedPeer: CmxIrohAdmittedPeer, artifactRegistry: MobileHostIrohArtifactTransferRegistry, - journal: IrxJournal + journal: IrxJournal, + onInteractiveSurface: @escaping MobileHostIrxTerminalLaneServer.InteractiveSurfaceObserver ) async { let terminalLaneQuota = MobileHostIrxTerminalLaneQuota() while !Task.isCancelled { @@ -1057,7 +1063,8 @@ final class MobileHostIrxRuntime: MobileHostPairingRuntime { resourceID: resource, cursor: cursor, stream: lane.bidirectional(), - journal: journal + journal: journal, + onInteractiveSurface: onInteractiveSurface ) await terminalLaneQuota.release() } @@ -1072,7 +1079,8 @@ final class MobileHostIrxRuntime: MobileHostPairingRuntime { await MobileHostIrxTerminalLaneServer.serveInputOnly( resourceID: resource, stream: lane.bidirectional(), - journal: journal + journal: journal, + onInteractiveSurface: onInteractiveSurface ) await terminalLaneQuota.release() } diff --git a/Sources/Mobile/MobileHostIrxTerminalLaneServer.swift b/Sources/Mobile/MobileHostIrxTerminalLaneServer.swift index 44cefd69622d..5b0ef74a8471 100644 --- a/Sources/Mobile/MobileHostIrxTerminalLaneServer.swift +++ b/Sources/Mobile/MobileHostIrxTerminalLaneServer.swift @@ -17,11 +17,16 @@ enum MobileHostIrxTerminalLaneServer { private static let maximumInputBufferByteCount = 64 * 1_024 + /// Called with the surface that received input, so the host can schedule + /// that surface's output stream first (keystroke echo). + typealias InteractiveSurfaceObserver = @Sendable (UUID) async -> Void + static func serve( resourceID: String, cursor: UInt64?, stream: CmxIrohBidirectionalStream, - journal: IrxJournal + journal: IrxJournal, + onInteractiveSurface: @escaping InteractiveSurfaceObserver = { _ in } ) async { guard let surfaceID = terminalSurfaceID(resourceID), await MainActor.run(body: { @@ -45,7 +50,11 @@ enum MobileHostIrxTerminalLaneServer { return true } group.addTask { - await receiveInput(surfaceID: surfaceID, stream: stream) + await receiveInput( + surfaceID: surfaceID, + stream: stream, + onInteractiveSurface: onInteractiveSurface + ) } if await group.next() == true { group.cancelAll() @@ -64,7 +73,8 @@ enum MobileHostIrxTerminalLaneServer { static func serveInputOnly( resourceID: String, stream: CmxIrohBidirectionalStream, - journal: IrxJournal + journal: IrxJournal, + onInteractiveSurface: @escaping InteractiveSurfaceObserver = { _ in } ) async { guard let surfaceID = terminalSurfaceID(resourceID), await MainActor.run(body: { @@ -88,9 +98,12 @@ enum MobileHostIrxTerminalLaneServer { try await stream.sendStream.send( CmxIrohTerminalOutputEnvelopeCodec().encode(baseline) ) + // The phone keeps an input lane for the terminal it shows. + await onInteractiveSurface(surfaceID) _ = await receiveInput( surfaceID: surfaceID, - stream: stream + stream: stream, + onInteractiveSurface: onInteractiveSurface ) } catch is CancellationError { await stream.sendStream.reset(errorCode: 0) @@ -206,7 +219,8 @@ enum MobileHostIrxTerminalLaneServer { /// on a clean input-side finish (output-only lanes stay open). private static func receiveInput( surfaceID: UUID, - stream: CmxIrohBidirectionalStream + stream: CmxIrohBidirectionalStream, + onInteractiveSurface: InteractiveSurfaceObserver ) async -> Bool { var buffer = Data() do { @@ -223,6 +237,7 @@ enum MobileHostIrxTerminalLaneServer { } for input in try MobileTerminalInputFrame.decode(from: &buffer) { + await onInteractiveSurface(surfaceID) guard await deliverInput( input, surfaceID: surfaceID diff --git a/Sources/Mobile/MobileHostService.swift b/Sources/Mobile/MobileHostService.swift index f6cf47d0c957..e10c7671b001 100644 --- a/Sources/Mobile/MobileHostService.swift +++ b/Sources/Mobile/MobileHostService.swift @@ -3,6 +3,7 @@ import CMUXMobileCore import CmuxAuthRuntime import CmuxGit import CmuxIrohTransport +import CmuxIrxTransport import CmuxMobileHost import CmuxMobileTransport import CmuxSettings @@ -664,7 +665,8 @@ final class MobileHostService { } resyncSurfaceIDs.formUnion(result.renderGridResyncSurfaceIDs) if result.startDrain { - Task { await connection.drainQueuedEvents() } + let lane = result.drainLane + Task { await connection.drainQueuedEvents(lane: lane) } } } @@ -1399,6 +1401,8 @@ actor MobileHostConnection { let topics: Set let transport: MobileHostEventTransport let clientID: String? + /// Render-grid frames for this stream ride per-surface lanes. + var surfaceEventLanes = false } private struct ResponseTask: Sendable { @@ -1460,6 +1464,12 @@ actor MobileHostConnection { private var receiveTask: Task? private var independentEventRevision: UInt64 = 0 private var independentEventNegotiationInProgress = false + /// Whether the event queue currently routes render-grid frames onto + /// per-surface lanes (mirrors the subscriptions that negotiated them). + private var surfaceEventLanesActive = false + /// Last surface this connection wrote terminal input to; its output lane + /// is scheduled first so keystroke echo never waits behind other surfaces. + private var lastInteractiveSurfaceKey: String? private var didDecodeFirstFrame = false private var isClosed = false private var exit = CmxIrohAdmittedConnectionExit( @@ -1705,6 +1715,7 @@ actor MobileHostConnection { if case let .success(request) = decodedRequest, request.isOrderedTerminalInput { let surfaceKey = request.orderedInputSurfaceKey + noteInteractiveSurface(surfaceKey) orderedRequestQueuesBySurfaceKey[surfaceKey, default: MobileHostOrderedRequestQueue()] .enqueue(MobileHostOrderedRequest( frameByteCount: frame.count, @@ -1973,11 +1984,17 @@ actor MobileHostConnection { } else { selectedTransport = .control } + let grantsSurfaceEventLanes = selectedTransport == .irohServerEvents + && topics.contains(MobileHostEventTopicPolicy.renderGridTopic) + && request.params[IrxSurfaceEventLaneProtocol().subscribeParameterKey] as? String + == IrxSurfaceEventLaneProtocol().subscribeParameterValue + && (independentEventWriter?.maximumSurfaceEventLaneCount ?? 0) > 0 await subscribe( streamID: streamID, topics: topics, transport: selectedTransport, - clientID: request.params["client_id"] as? String + clientID: request.params["client_id"] as? String, + surfaceEventLanes: grantsSurfaceEventLanes ) if topics.contains("terminal.render_grid") { // Anchor negotiation: "screen" clients own their local @@ -1993,12 +2010,17 @@ actor MobileHostConnection { #if DEBUG cmuxDebugLog("mobile.subscribe streamID=\(streamID) topics=\(topics.sorted()) existing=\(alreadySubscribed) connID=\(self.id.uuidString)") #endif - return .ok([ + var acknowledgement: [String: Any] = [ "stream_id": streamID, "topics": Array(topics).sorted(), "already_subscribed": alreadySubscribed, "event_transport": selectedTransport.rawValue, - ]) + ] + if subscriptions[streamID]?.surfaceEventLanes == true { + acknowledgement[IrxSurfaceEventLaneProtocol().subscribeParameterKey] = + IrxSurfaceEventLaneProtocol().subscribeParameterValue + } + return .ok(acknowledgement) case "mobile.events.unsubscribe": let streamID = request.params["stream_id"] as? String ?? "" let removed = await unsubscribe(streamID: streamID) @@ -2115,13 +2137,15 @@ actor MobileHostConnection { streamID: String, topics: Set, transport: MobileHostEventTransport = .control, - clientID: String? = nil + clientID: String? = nil, + surfaceEventLanes: Bool = false ) async { let previousTopics = subscriptions[streamID]?.topics subscriptions[streamID] = EventSubscription( topics: topics, transport: transport, - clientID: clientID + clientID: clientID, + surfaceEventLanes: surfaceEventLanes ) if let usableEventSubscription, usableEventSubscription.streamID == streamID, @@ -2133,6 +2157,7 @@ actor MobileHostConnection { previousTopics: previousTopics, nextTopics: topics ) + await syncSurfaceEventLanes() if currentSubscribedTopics().contains(MobileHostEventTopicPolicy.simulatorFrameTopic) { await dispatchPendingSimulatorFrameReplay() } @@ -2153,6 +2178,7 @@ actor MobileHostConnection { nextTopics: nil ) } + await syncSurfaceEventLanes() if !subscriptions.values.contains(where: { $0.transport == .irohServerEvents }) { @@ -2216,7 +2242,8 @@ actor MobileHostConnection { ) } if result.startDrain { - Task { await self.drainQueuedEvents() } + let lane = result.drainLane + Task { await self.drainQueuedEvents(lane: lane) } } return result.admitted } @@ -2253,8 +2280,8 @@ actor MobileHostConnection { independentEventNegotiationInProgress = true defer { independentEventNegotiationInProgress = false - if eventQueue.claimDrain() { - Task { await self.drainQueuedEvents() } + for lane in eventQueue.claimDrains() { + Task { await self.drainQueuedEvents(lane: lane) } } } let probePayload = Data(#"{"kind":"event_stream_probe"}"#.utf8) @@ -2274,22 +2301,25 @@ actor MobileHostConnection { await resetIndependentEventWriter() } downgradeIndependentSubscriptionsToControl() + await syncSurfaceEventLanes() return false } - /// Single-writer drain loop: at most one instance runs per connection - /// (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 (which closes the unusable control session). - func drainQueuedEvents() async { + /// Single-writer drain loop per lane: at most one instance runs per lane + /// (enforced by the queue's drain claim), pulling that lane's events from + /// the bounded queue and writing them to the lane's stream. Lanes drain + /// independently, so a write stalled on one surface's stream never delays + /// another lane. Exits when the lane is empty, the connection closes, lane + /// negotiation pauses delivery, or a delivery fails (which closes the + /// unusable control session). + func drainQueuedEvents(lane: MobileHostEventLane = .shared) async { while true { if isClosed || independentEventNegotiationInProgress { - eventQueue.abandonDrain() + eventQueue.abandonDrain(lane: lane) return } - guard let event = eventQueue.dequeue() else { - if eventQueue.finishDrain() { continue } + guard let event = eventQueue.dequeue(lane: lane) else { + if eventQueue.finishDrain(lane: lane) { continue } return } guard eventQueue.isSubscribed(topic: event.topic) else { continue } @@ -2297,7 +2327,7 @@ actor MobileHostConnection { let latencyWriteStart = event.stateSeq == nil ? nil : HostLatencyTrace.captureTime() #endif guard await deliverQueuedEvent(event) else { - eventQueue.abandonDrain() + eventQueue.abandonDrain(lane: lane) return } #if DEBUG @@ -2341,6 +2371,10 @@ actor MobileHostConnection { } private func deliverQueuedEvent(_ event: MobileHostConnectionEventQueue.QueuedEvent) async -> Bool { + if case .surface(let surfaceID) = event.lane { + await deliverSurfaceEvent(event, surfaceID: surfaceID) + return true + } let prefersIndependent = subscriptions.values.contains { $0.transport == .irohServerEvents && $0.topics.contains(event.topic) } @@ -2351,6 +2385,7 @@ actor MobileHostConnection { } catch { independentEventRevision &+= 1 downgradeIndependentSubscriptionsToControl() + await syncSurfaceEventLanes() await independentEventWriter.reset() // Deliver the event that exposed the dead/backpressured lane on // control immediately. Subsequent events also use control. @@ -2360,6 +2395,82 @@ actor MobileHostConnection { return await sendEventControlFrame(event.frame) } + /// Writes one render-grid frame onto its surface's own stream. A failed or + /// stalled stream is retired for that surface only: its lost frames break + /// the surface's chain, so the queue poisons it and the producer re-emits a + /// full frame, which opens a fresh stream. The connection stays up. + private func deliverSurfaceEvent( + _ event: MobileHostConnectionEventQueue.QueuedEvent, + surfaceID: String + ) async { + guard let independentEventWriter else { return } + do { + try await independentEventWriter.sendSurfaceEvent( + event.frame, + surfaceID: surfaceID, + generation: event.laneGeneration + ) + eventQueue.noteSurfaceLaneDelivered(surfaceID: surfaceID) + } catch { + let resync = eventQueue.retireSurfaceLane( + surfaceID: surfaceID, + generation: event.laneGeneration + ) + mobileHostLog.info( + "mobile host retired surface event lane \(surfaceID, privacy: .public): \(String(describing: error), privacy: .public)" + ) + if !resync.isEmpty { + MobileTerminalRenderObserver.requestRenderGridFullResync( + surfaceIDStrings: resync + ) + } + } + } + + /// Aligns the queue's render-grid routing and the writer's surface lanes + /// with the subscriptions that negotiated them. Surface lanes stay active + /// only while a render-grid subscription still uses the independent + /// events path; a fallback to control returns every surface to the shared + /// lane and re-bases each chain with a full frame. + private func syncSurfaceEventLanes() async { + guard let independentEventWriter, !isClosed else { return } + let desired = subscriptions.values.contains { + $0.surfaceEventLanes + && $0.transport == .irohServerEvents + && $0.topics.contains(MobileHostEventTopicPolicy.renderGridTopic) + } + guard desired != surfaceEventLanesActive else { return } + // Flip the flag and the queue's routing before any suspension so a + // reentrant sync always observes and applies the latest decision; the + // writer's own enable calls reach its actor in the same order. + surfaceEventLanesActive = desired + if desired { + eventQueue.enableSurfaceLanes( + limit: independentEventWriter.maximumSurfaceEventLaneCount + ) + let focusedSurfaceKey = lastInteractiveSurfaceKey + await independentEventWriter.setSurfaceEventLanesEnabled(true) + if let focusedSurfaceKey { + await independentEventWriter.noteInteractiveSurface(focusedSurfaceKey) + } + } else { + let resync = eventQueue.disableSurfaceLanes() + if !resync.isEmpty { + MobileTerminalRenderObserver.requestRenderGridFullResync( + surfaceIDStrings: resync + ) + } + await independentEventWriter.setSurfaceEventLanesEnabled(false) + } + } + + private func noteInteractiveSurface(_ surfaceKey: String) { + guard !surfaceKey.isEmpty, lastInteractiveSurfaceKey != surfaceKey else { return } + lastInteractiveSurfaceKey = surfaceKey + guard surfaceEventLanesActive, let independentEventWriter else { return } + Task { await independentEventWriter.noteInteractiveSurface(surfaceKey) } + } + /// 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. @@ -2373,7 +2484,8 @@ actor MobileHostConnection { subscriptions[streamID] = EventSubscription( topics: subscription.topics, transport: .control, - clientID: subscription.clientID + clientID: subscription.clientID, + surfaceEventLanes: false ) } if let usableEventSubscription, diff --git a/Sources/Mobile/MobileHostTransportAuthorization.swift b/Sources/Mobile/MobileHostTransportAuthorization.swift index 679ae3702d93..abeee2894442 100644 --- a/Sources/Mobile/MobileHostTransportAuthorization.swift +++ b/Sources/Mobile/MobileHostTransportAuthorization.swift @@ -57,6 +57,30 @@ protocol MobileHostIndependentEventWriting: Sendable { func send(_ framedData: Data) async throws func reset() async func close() async + + /// Surface lanes put each terminal's render-grid frames on its own QUIC + /// stream so one terminal's burst cannot head-of-line-block another's. + /// Only a writer that owns a multi-stream connection supports them. + var maximumSurfaceEventLaneCount: Int { get } + /// Writes one frame onto the surface's own stream. A throw means that + /// stream was retired; the next send for a newer `generation` reopens. + func sendSurfaceEvent(_ framedData: Data, surfaceID: String, generation: UInt64) async throws + /// Enables or disables surface lanes; disabling finishes every lane. + func setSurfaceEventLanesEnabled(_ enabled: Bool) async + /// Raises the stream priority of the surface the user is interacting with. + func noteInteractiveSurface(_ surfaceID: String) async +} + +extension MobileHostIndependentEventWriting { + var maximumSurfaceEventLaneCount: Int { 0 } + + func sendSurfaceEvent(_ framedData: Data, surfaceID _: String, generation _: UInt64) async throws { + try await send(framedData) + } + + func setSurfaceEventLanesEnabled(_: Bool) async {} + + func noteInteractiveSurface(_: String) async {} } final class MobileHostConnectionRegistry: @unchecked Sendable { diff --git a/cmux.xcodeproj/project.pbxproj b/cmux.xcodeproj/project.pbxproj index 80dd6dfe51e2..8cb00f8c574c 100644 --- a/cmux.xcodeproj/project.pbxproj +++ b/cmux.xcodeproj/project.pbxproj @@ -2260,6 +2260,7 @@ C0DE10510000000000000001 /* MobileHostServiceSettingsTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0DE10510000000000000002 /* MobileHostServiceSettingsTests.swift */; }; 4C1A7E10B2D34F56A8C90011 /* MobileHostStatusVerificationLimiter.swift in Sources */ = {isa = PBXBuildFile; fileRef = 4C1A7E10B2D34F56A8C90012 /* MobileHostStatusVerificationLimiter.swift */; }; 4C1A7E10B2D34F56A8C90013 /* MobileHostStatusVerificationLimiterTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 4C1A7E10B2D34F56A8C90014 /* MobileHostStatusVerificationLimiterTests.swift */; }; + A2111A137939488266F270C5 /* MobileHostSurfaceEventLaneTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 1EC28906670BF54F49F89FFE /* MobileHostSurfaceEventLaneTests.swift */; }; C7A50B000000000000000020 /* MobileHostTerminalTheme.swift in Sources */ = {isa = PBXBuildFile; fileRef = C7A50B000000000000000021 /* MobileHostTerminalTheme.swift */; }; 7A1B2C3D4E5F60718293A4B6 /* MobileHostTerminalThemeTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7A1B2C3D4E5F60718293A4B5 /* MobileHostTerminalThemeTests.swift */; }; C1A070000000000000000005 /* MobileHostTransportAuthorization.swift in Sources */ = {isa = PBXBuildFile; fileRef = C1A070000000000000000015 /* MobileHostTransportAuthorization.swift */; }; @@ -6221,6 +6222,7 @@ C0DE10510000000000000002 /* MobileHostServiceSettingsTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MobileHostServiceSettingsTests.swift; sourceTree = ""; }; 4C1A7E10B2D34F56A8C90012 /* MobileHostStatusVerificationLimiter.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = MobileHostStatusVerificationLimiter.swift; sourceTree = ""; }; 4C1A7E10B2D34F56A8C90014 /* MobileHostStatusVerificationLimiterTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MobileHostStatusVerificationLimiterTests.swift; sourceTree = ""; }; + 1EC28906670BF54F49F89FFE /* MobileHostSurfaceEventLaneTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "MobileHostSurfaceEventLaneTests.swift"; sourceTree = ""; }; C7A50B000000000000000021 /* MobileHostTerminalTheme.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = MobileHostTerminalTheme.swift; sourceTree = ""; }; 7A1B2C3D4E5F60718293A4B5 /* MobileHostTerminalThemeTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = MobileHostTerminalThemeTests.swift; sourceTree = ""; }; C1A070000000000000000015 /* MobileHostTransportAuthorization.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = MobileHostTransportAuthorization.swift; sourceTree = ""; }; @@ -12089,6 +12091,7 @@ E97B65696F71CE797D211070 /* ScrollbackTestTerminal.swift */, B5C6895DE9644DD1B59FDCE2 /* SidebarWorkspaceDragDelegateTests.swift */, F5C553AF2050E7E6D7CD62EC /* VMSSHTransportFallbackTests.swift */, + 1EC28906670BF54F49F89FFE /* MobileHostSurfaceEventLaneTests.swift */, CDB53053B8C69BDF0B219DE3 /* CloudWorkspaceRenameRefreshTests.swift */, C1A102DC561C2D349FD9A09C /* ComputerUseOnboardingIntentTests.swift */, ); @@ -16234,6 +16237,7 @@ A1B2C3D4E5F60718293A4C03 /* MobileHostOrderedInputTests.swift in Sources */, C0DE10510000000000000001 /* MobileHostServiceSettingsTests.swift in Sources */, 4C1A7E10B2D34F56A8C90013 /* MobileHostStatusVerificationLimiterTests.swift in Sources */, + A2111A137939488266F270C5 /* MobileHostSurfaceEventLaneTests.swift in Sources */, 7A1B2C3D4E5F60718293A4B6 /* MobileHostTerminalThemeTests.swift in Sources */, C0DE73840000000000000001 /* MobileHostWorkspaceTicketAuthorizationTests.swift in Sources */, 9B08F916D7FF7626C6820727 /* MobilePairingConnectionTransitionTests.swift in Sources */, diff --git a/cmuxTests/MobileHostSurfaceEventLaneTests.swift b/cmuxTests/MobileHostSurfaceEventLaneTests.swift new file mode 100644 index 000000000000..89da95cedfbb --- /dev/null +++ b/cmuxTests/MobileHostSurfaceEventLaneTests.swift @@ -0,0 +1,345 @@ +import CMUXMobileCore +import CmuxIrohTransport +import Foundation +import Testing + +#if canImport(cmux_DEV) +@testable import cmux_DEV +#elseif canImport(cmux) +@testable import cmux +#endif + +/// One terminal's render-grid output must never wait behind another +/// terminal's output on the phone connection (QUIC head-of-line blocking). +@Suite(.serialized) +struct MobileHostSurfaceEventLaneTests { + @Test func stalledSurfaceOutputDoesNotDelayAnotherSurfacesRenderGrid() async throws { + let control = RecordingMobileHostByteTransport() + let writer = SurfaceGatedIndependentEventWriter(stalledSurfaceID: "surface-a") + let session = MobileHostConnection( + id: UUID(), + transport: control, + independentEventWriter: writer, + authorizeRequest: { _ in nil }, + onAuthorizedRequest: { _ in }, + handleRequest: { _ in .ok([:]) }, + onClose: { _ in } + ) + _ = await session.debugHandleSubscriptionRPCForTesting( + MobileHostRPCRequest( + id: "subscribe", + method: "mobile.events.subscribe", + params: [ + "stream_id": "events", + "topics": ["terminal.render_grid"], + "event_transport": "iroh_server_events_v1", + "surface_event_lanes": "v1", + ], + auth: nil + ) + ) + + // Surface A's replay is stuck in flight: the phone has not granted + // stream credit for its bytes yet. + #expect(await session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-a", "full": true, "rows": ["replay"]] + )) + #expect(await writer.waitUntilStalled()) + + // The user types on surface B. Its echo must reach the wire now, not + // after surface A's replay drains. + #expect(await session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-b", "full": true, "rows": ["echo"]] + )) + let delivered = await writer.waitForDeliveredSurface("surface-b") + #expect(delivered, "surface-b's render grid waited behind surface-a's stalled write") + + await session.close(reason: "test complete") + } + + @Test func hostGrantsSurfaceLanesOnlyWhenThePhoneAsksForThem() async throws { + for asks in [false, true] { + let writer = RecordingSurfaceLaneEventWriter() + let session = MobileHostConnection( + id: UUID(), + transport: RecordingMobileHostByteTransport(), + independentEventWriter: writer, + authorizeRequest: { _ in nil }, + onAuthorizedRequest: { _ in }, + handleRequest: { _ in .ok([:]) }, + onClose: { _ in } + ) + var params: [String: Any] = [ + "stream_id": "events", + "topics": ["terminal.render_grid"], + "event_transport": "iroh_server_events_v1", + ] + if asks { params["surface_event_lanes"] = "v1" } + let result = await session.debugHandleSubscriptionRPCForTesting( + MobileHostRPCRequest(id: "s", method: "mobile.events.subscribe", params: params, auth: nil) + ) + guard case let .ok(payload)? = result, + let acknowledgement = payload as? [String: Any] else { + Issue.record("Expected a subscribe acknowledgement") + return + } + // An older phone reads exactly one events stream, so it must never + // be sent a second one. + #expect((acknowledgement["surface_event_lanes"] as? String) == (asks ? "v1" : nil)) + + #expect(await session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-a", "full": true] + )) + #expect(await writer.waitForWrites(1)) + #expect(await writer.writes() == [asks ? .surface("surface-a", generation: 0) : .shared]) + await session.close(reason: "test complete") + } + } + + @Test func failedSurfaceLaneRecoversOnAFreshStreamWithoutClosingTheConnection() async throws { + let writer = RecordingSurfaceLaneEventWriter(failFirstSurfaceWrite: true) + let recorder = MobileHostConnectionCloseRecorder() + let session = MobileHostConnection( + id: UUID(), + transport: RecordingMobileHostByteTransport(), + independentEventWriter: writer, + authorizeRequest: { _ in nil }, + onAuthorizedRequest: { _ in }, + handleRequest: { _ in .ok([:]) }, + onClose: { id in await recorder.record(id) } + ) + _ = await session.debugHandleSubscriptionRPCForTesting( + MobileHostRPCRequest( + id: "s", + method: "mobile.events.subscribe", + params: [ + "stream_id": "events", + "topics": ["terminal.render_grid"], + "event_transport": "iroh_server_events_v1", + "surface_event_lanes": "v1", + ], + auth: nil + ) + ) + #expect(await session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-a", "full": true] + )) + #expect(await writer.waitForWrites(1)) + for _ in 0..<2_000 where session.eventQueue.surfaceLaneGeneration(surfaceID: "surface-a") == 0 { + try? await Task.sleep(nanoseconds: 1_000_000) + } + #expect(session.eventQueue.surfaceLaneGeneration(surfaceID: "surface-a") == 1) + // The lost frame broke the chain: deltas are refused until the + // producer's full-frame resync, which rides a new stream generation. + #expect(await !session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-a", "full": false] + )) + #expect(await session.sendEvent( + topic: "terminal.render_grid", + payload: ["surface_id": "surface-a", "full": true] + )) + #expect(await writer.waitForWrites(2)) + #expect(await writer.writes() == [ + .surface("surface-a", generation: 0), + .surface("surface-a", generation: 1), + ]) + #expect(await recorder.recordedIDs().isEmpty) + await session.close(reason: "test complete") + } + + @Test func surfaceLaneFallbackToControlRebasesEachSurfaceWithAFullFrame() { + let queue = MobileHostConnectionEventQueue() + queue.updateSubscribedTopics(["terminal.render_grid"]) + queue.enableSurfaceLanes(limit: 4) + let full = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: true, frame: Data([1]) + ) + #expect(full.drainLane == .surface("s")) + let delta = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: false, frame: Data([2]) + ) + #expect(delta.admitted) + + // Falling back drops the frames still queued for the surface stream + // and poisons the surface until a full frame on the shared lane. + #expect(queue.disableSurfaceLanes() == ["s"]) + #expect(queue.count == 0) + let staleDelta = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: false, frame: Data([3]) + ) + #expect(!staleDelta.admitted) + let rebase = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: true, frame: Data([4]) + ) + #expect(rebase.admitted) + #expect(rebase.drainLane == .shared) + } + + @Test func deltaCrossingOntoANewStreamRequestsAFullFrame() { + let queue = MobileHostConnectionEventQueue() + queue.updateSubscribedTopics(["terminal.render_grid"]) + _ = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: true, frame: Data([1]) + ) + // Lanes are negotiated mid-chain: a delta may not follow its base + // onto a different stream, where it could overtake it. + queue.enableSurfaceLanes(limit: 4) + let crossing = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: false, frame: Data([2]) + ) + #expect(!crossing.admitted) + #expect(crossing.renderGridResyncSurfaceIDs == ["s"]) + let rebase = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: true, frame: Data([3]) + ) + #expect(rebase.drainLane == .surface("s")) + let next = queue.enqueue( + topic: "terminal.render_grid", coalesceKey: "s", isFullRenderGridFrame: false, frame: Data([4]) + ) + #expect(next.admitted) + #expect(!next.startDrain) + } + + @Test func eachSurfaceLaneDrainsIndependentlyAndOverflowRidesTheSharedLane() { + let queue = MobileHostConnectionEventQueue() + queue.updateSubscribedTopics(["terminal.render_grid", "workspace.updated"]) + queue.enableSurfaceLanes(limit: 2) + let a = queue.enqueue(topic: "terminal.render_grid", coalesceKey: "a", isFullRenderGridFrame: true, frame: Data([1])) + let b = queue.enqueue(topic: "terminal.render_grid", coalesceKey: "b", isFullRenderGridFrame: true, frame: Data([2])) + let c = queue.enqueue(topic: "terminal.render_grid", coalesceKey: "c", isFullRenderGridFrame: true, frame: Data([3])) + let shared = queue.enqueue(topic: "workspace.updated", coalesceKey: nil, isFullRenderGridFrame: false, frame: Data([4])) + #expect(a.startDrain && a.drainLane == .surface("a")) + #expect(b.startDrain && b.drainLane == .surface("b")) + // Both surface lanes are busy, so the bounded third surface shares. + #expect(c.startDrain && c.drainLane == .shared) + #expect(!shared.startDrain) + #expect(queue.dequeue(lane: .surface("b"))?.frame == Data([2])) + #expect(queue.dequeue(lane: .shared)?.frame == Data([3])) + #expect(queue.dequeue(lane: .shared)?.frame == Data([4])) + #expect(queue.dequeue(lane: .surface("a"))?.frame == Data([1])) + } + + @Test func repeatedSurfaceLaneFailuresPinTheSurfaceToTheSharedLane() { + let queue = MobileHostConnectionEventQueue() + queue.updateSubscribedTopics(["terminal.render_grid"]) + queue.enableSurfaceLanes(limit: 4) + for attempt in 0.. Bool { true } + func send(_: Data) async throws { recorded.append(.shared) } + func reset() async {} + func close() async {} + + func sendSurfaceEvent(_: Data, surfaceID: String, generation: UInt64) async throws { + recorded.append(.surface(surfaceID, generation: generation)) + if failFirstSurfaceWrite { + failFirstSurfaceWrite = false + throw CancellationError() + } + } + + func writes() -> [Write] { recorded } + + func waitForWrites(_ count: Int) async -> Bool { + for _ in 0..<2_000 { + if recorded.count >= count { return true } + try? await Task.sleep(nanoseconds: 1_000_000) + } + return false + } +} + +/// Independent event writer whose writes for one surface never complete +/// until the connection closes, like a QUIC stream out of flow credit. +actor SurfaceGatedIndependentEventWriter: MobileHostIndependentEventWriting { + nonisolated let maximumSurfaceEventLaneCount = 16 + private let stalledSurfaceID: String + private var stalledWrites: [CheckedContinuation] = [] + private var deliveredSurfaceIDs: [String] = [] + + init(stalledSurfaceID: String) { + self.stalledSurfaceID = stalledSurfaceID + } + + func probe(_: Data) async -> Bool { true } + + func send(_ framedData: Data) async throws { + let surfaceID = Self.surfaceID(in: framedData) + if surfaceID == stalledSurfaceID { + try await withCheckedThrowingContinuation { stalledWrites.append($0) } + } + if let surfaceID { deliveredSurfaceIDs.append(surfaceID) } + } + + func reset() async {} + + func close() async { + let writes = stalledWrites + stalledWrites.removeAll() + for write in writes { write.resume(throwing: CancellationError()) } + } + + func waitUntilStalled() async -> Bool { + for _ in 0..<2_000 { + if !stalledWrites.isEmpty { return true } + try? await Task.sleep(nanoseconds: 1_000_000) + } + return false + } + + func waitForDeliveredSurface(_ surfaceID: String) async -> Bool { + for _ in 0..<2_000 { + if deliveredSurfaceIDs.contains(surfaceID) { return true } + try? await Task.sleep(nanoseconds: 1_000_000) + } + return false + } + + func delivered() -> [String] { deliveredSurfaceIDs } + + private static func surfaceID(in framedData: Data) -> String? { + var buffer = framedData + guard let payload = try? MobileSyncFrameCodec.decodeFrames(from: &buffer).first, + let envelope = try? JSONSerialization.jsonObject(with: payload) as? [String: Any], + let body = envelope["payload"] as? [String: Any] else { return nil } + return body["surface_id"] as? String + } +} diff --git a/ios/cmux/cmuxApp.swift b/ios/cmux/cmuxApp.swift index 101096b44d75..458472af1b04 100644 --- a/ios/cmux/cmuxApp.swift +++ b/ios/cmux/cmuxApp.swift @@ -105,7 +105,9 @@ struct cmuxApp: App { simulatorStreamLaneProvider: { request, panelID in guard let panelUUID = UUID(uuidString: panelID) else { throw MobileIrohSimulatorStreamLaneError.invalidPanelID } return try await irx.openSimulatorStreamLane(for: request, panelID: panelUUID) - } + }, + // irx.serverEventByteStream merges every per-surface event lane. + independentEventsMergeSurfaceLanes: true ) return AppCompositionRoot( diff --git a/ios/cmuxPackage/Sources/cmuxFeature/CMUXMobileRuntime.swift b/ios/cmuxPackage/Sources/cmuxFeature/CMUXMobileRuntime.swift index 9f943af06e5a..ccb5566ad77e 100644 --- a/ios/cmuxPackage/Sources/cmuxFeature/CMUXMobileRuntime.swift +++ b/ios/cmuxPackage/Sources/cmuxFeature/CMUXMobileRuntime.swift @@ -34,6 +34,8 @@ public struct CMUXMobileRuntime: Sendable, MobileSyncRuntime { /// 750ms poll only when a connected Mac does not support events. public var supportsServerPushEvents: Bool public var independentEventByteStreamProvider: CmxIndependentEventByteStreamProvider? + /// Set when the provider merges per-surface event lanes (the irx hub). + public var independentEventsMergeSurfaceLanes: Bool public var terminalLaneProvider: MobileTerminalLaneProvider? public var terminalInputLaneProvider: MobileTerminalLaneProvider? public var simulatorStreamLaneProvider: MobileSimulatorStreamLaneProvider? @@ -149,7 +151,8 @@ public struct CMUXMobileRuntime: Sendable, MobileSyncRuntime { terminalLaneProvider: MobileTerminalLaneProvider? = nil, terminalInputLaneProvider: MobileTerminalLaneProvider? = nil, artifactLaneProvider: MobileArtifactLaneProvider? = nil, - simulatorStreamLaneProvider: MobileSimulatorStreamLaneProvider? = nil + simulatorStreamLaneProvider: MobileSimulatorStreamLaneProvider? = nil, + independentEventsMergeSurfaceLanes: Bool = false ) { self.supportedRouteKinds = supportedRouteKinds self.transportFactory = transportFactory @@ -166,6 +169,7 @@ public struct CMUXMobileRuntime: Sendable, MobileSyncRuntime { self.terminalInputLaneProvider = terminalInputLaneProvider self.artifactLaneProvider = artifactLaneProvider self.simulatorStreamLaneProvider = simulatorStreamLaneProvider + self.independentEventsMergeSurfaceLanes = independentEventsMergeSurfaceLanes } public init( @@ -182,7 +186,8 @@ public struct CMUXMobileRuntime: Sendable, MobileSyncRuntime { terminalLaneProvider: MobileTerminalLaneProvider? = nil, terminalInputLaneProvider: MobileTerminalLaneProvider? = nil, artifactLaneProvider: MobileArtifactLaneProvider? = nil, - simulatorStreamLaneProvider: MobileSimulatorStreamLaneProvider? = nil + simulatorStreamLaneProvider: MobileSimulatorStreamLaneProvider? = nil, + independentEventsMergeSurfaceLanes: Bool = false ) { self.supportedRouteKinds = transportFactory.supportedKinds self.transportFactory = transportFactory @@ -198,6 +203,7 @@ public struct CMUXMobileRuntime: Sendable, MobileSyncRuntime { self.terminalInputLaneProvider = terminalInputLaneProvider self.artifactLaneProvider = artifactLaneProvider self.simulatorStreamLaneProvider = simulatorStreamLaneProvider + self.independentEventsMergeSurfaceLanes = independentEventsMergeSurfaceLanes self.now = now } } diff --git a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Dial.swift b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Dial.swift index f57f3874b85a..84f35437db4b 100644 --- a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Dial.swift +++ b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Dial.swift @@ -158,7 +158,10 @@ extension MobileIrxRuntimeComposition { try await assertScope(scope, epoch: currentEpoch) let (admit, control) = try await IrxAdmission().performClient(connection: connection, journal: journal) try await assertScope(scope, epoch: currentEpoch) - await connection.raiseRemoteStreamCredit(bi: 0, uni: 4) + // One shared events lane plus up to 16 per-terminal output lanes + // (IrxSurfaceEventLanes), with headroom for streams the Mac is + // replacing. An older Mac opens only the shared lane. + await connection.raiseRemoteStreamCredit(bi: 0, uni: 40) if !forceRelayOnly, case .automatic = intent { await connection.authorizeDirectPaths() } try await assertScope(scope, epoch: currentEpoch) activeDialIntentByPeer[peerHex] = intent diff --git a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Lifecycle.swift b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Lifecycle.swift index c5b0b302c76d..c224020d4ed7 100644 --- a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Lifecycle.swift +++ b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Lifecycle.swift @@ -277,6 +277,9 @@ extension MobileIrxRuntimeComposition { lastFailure = nil enginesByPeer.removeAll(); dialIntentByPeer.removeAll(); activeDialIntentByPeer.removeAll() expectedDeviceIDByPeer.removeAll(); controlLaneClaims.removeAll(); claimedEventSessions.removeAll() + let oldEventLaneHubs = eventLaneHubs.values.map(\.hub) + eventLaneHubs.removeAll() + for hub in oldEventLaneHubs { Task { await hub.stop() } } publish() await MainActor.run { self.macListAuthState.clear() } return DetachedRuntime( diff --git a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Streams.swift b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Streams.swift index 3e424e819eb3..45b585aeed37 100644 --- a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Streams.swift +++ b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition+Streams.swift @@ -13,16 +13,14 @@ extension MobileIrxRuntimeComposition { throw IrxConnectionError.closed(nil) } claimedEventSessions[peerHex] = session.admit.session - let connection = session.connection - let journal = journal + // The hub reads the shared events lane and every per-surface lane the + // Mac opens, each on its own task, and forwards whole frames. One + // terminal's burst therefore never delays another terminal's frames. + let merged = await eventLaneHub(peerHex: peerHex, session: session).subscribe() return AsyncThrowingStream { continuation in let pump = Task { do { - guard let (descriptor, reader) = try await connection.acceptUniLane(), descriptor.lane == .events else { - throw IrxConnectionError.closed(nil) - } - journal.record("client-events", "lane-accepted") - while let chunk = try await reader.readRaw() { + for try await chunk in merged { guard !Task.isCancelled else { throw CancellationError() } continuation.yield(chunk) } @@ -43,6 +41,26 @@ extension MobileIrxRuntimeComposition { } } + /// Returns the session's lane hub, replacing one left by an older session. + /// The hub owns uni-lane acceptance for the connection's whole lifetime, + /// so a replaced reader never leaves an accept loop that steals lanes. + func eventLaneHub(peerHex: String, session: IrxClientSession) async -> IrxServerEventLaneHub { + if let existing = eventLaneHubs[peerHex] { + if existing.sessionID == session.admit.session, await existing.hub.isAlive { + return existing.hub + } + let stale = existing.hub + Task { await stale.stop() } + } + let connection = session.connection + let hub = IrxServerEventLaneHub(journal: journal) { + guard let (descriptor, reader) = try await connection.acceptUniLane() else { return nil } + return (descriptor, reader) + } + eventLaneHubs[peerHex] = (session.admit.session, hub) + return hub + } + func releaseEventClaim(peerHex: String, sessionID: String) { guard claimedEventSessions[peerHex] == sessionID else { return } claimedEventSessions.removeValue(forKey: peerHex) diff --git a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition.swift b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition.swift index 0b6eaec0a4b4..7979b5de754c 100644 --- a/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition.swift +++ b/ios/cmuxPackage/Sources/cmuxFeature/MobileIrxRuntimeComposition.swift @@ -49,6 +49,8 @@ public actor MobileIrxRuntimeComposition { var expectedDeviceIDByPeer: [String: String] = [:] var controlLaneClaims = MobileIrxControlLaneClaims() var claimedEventSessions: [String: String] = [:] + /// One server-event lane acceptor per admitted session, keyed by peer. + var eventLaneHubs: [String: (sessionID: String, hub: IrxServerEventLaneHub)] = [:] var changeObservers: [UUID: AsyncStream.Continuation] = [:] var launchTime = Date() var backgroundTime: Date?