From 1827328ca44d5f4b8085470ce1a6fccca4e40544 Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 14:44:10 +0800 Subject: [PATCH 01/12] feat(apple): add scoped RVT1 client and gateway transport --- CHANGELOG.md | 9 + docs/desktop-closeout-mobile-plan.md | 8 + docs/device-communication.md | 12 +- docs/network-device-channel.md | 18 +- .../Sources/RivetHost/GeneratedBackend.swift | 4 +- platform/macos/Package.swift | 9 +- .../Sources/RivetDevice/DeviceGateway.swift | 61 ++- .../RivetDeviceAppleRVT1Channel.swift | 127 +++++++ .../RivetDeviceAppleRVT1Client.swift | 214 +++++++++++ .../RivetDeviceAppleRVT1GatewayServer.swift | 275 ++++++++++++++ .../RivetDeviceRVT1BackendHost.swift | 66 ++++ .../Sources/RivetRuntime/RivetRPCClient.swift | 21 ++ .../AppleRVT1GatewayTests.swift | 348 ++++++++++++++++++ .../RivetDeviceTests/DeviceGatewayTests.swift | 36 ++ .../RivetRPCClientTests.swift | 32 ++ .../Sources/RivetHost/GeneratedBackend.swift | 4 +- rivet-cli/codegen/swift.rkt | 2 +- rivet/tests/codegen.rkt | 2 + 18 files changed, 1224 insertions(+), 24 deletions(-) create mode 100644 platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Channel.swift create mode 100644 platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift create mode 100644 platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift create mode 100644 platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift create mode 100644 platform/macos/Sources/RivetRuntime/RivetRPCClient.swift create mode 100644 platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift create mode 100644 platform/macos/Tests/RivetRuntimeTests/RivetRPCClientTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index e2a6e54..297c5c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +- Add the Apple post-RVD1 RVT1 device path: generated Swift APIs can target a + shared RivetRPCClient protocol; a bounded network client correlates RPCs, + forwards cancellation and delivers events; a scoped gateway server applies + RPC/state/event grants and a backend adapter calls the existing local + client. Framework tests cover request correlation, state grant isolation, + event filtering and host argument/result encoding. Product listener wiring, + scope selection, event fan-out, signed apps and physical-device validation + remain outstanding. + - Bridge Android companion TLS streams to the RVD1 client through a one-shot selected-endpoint session driver. Enforce the stored SPKI pin on reconnect without implicit pairing fallback, preserve explicit pairing diff --git a/docs/desktop-closeout-mobile-plan.md b/docs/desktop-closeout-mobile-plan.md index 0527b37..c265a58 100644 --- a/docs/desktop-closeout-mobile-plan.md +++ b/docs/desktop-closeout-mobile-plan.md @@ -309,6 +309,14 @@ unverified. Add the Windows gateway against the same iOS client, then bind Android discovery/TLS streams to the existing Kotlin runtime and generated device API. +The Apple post-RVD1 RVT1 path now has a bounded client and scoped gateway +server, a backend adapter for the existing local RPC client, cancellation +forwarding, and per-device event/state-change filtering. The generated Swift +API accepts the shared RivetRPCClient protocol so in-process and network +clients can share typed methods. Product host wiring, per-device scope +selection, event fan-out, mobile UI integration, and real-device acceptance +remain open gates. + Apple companion coverage extends to iOS and iPadOS through the shared iOS transport, and watchOS through a separate WatchConnectivity request relay to its paired iPhone. The relay requires activated/reachable diff --git a/docs/device-communication.md b/docs/device-communication.md index 4fd61e0..02b04a2 100644 --- a/docs/device-communication.md +++ b/docs/device-communication.md @@ -107,11 +107,13 @@ pre-session admission and fail-closed lifecycle handling. That listener now publishes the privacy-bounded Bonjour service while ready. The framework-only Bonjour browser discovers that service on iOS, rejects non-canonical TXT records, and reports endpoint and advertisement changes without connecting or -granting authority. An iOS client-session coordinator and explicit native -confirmation UI remain. The macOS session host -drives `RVD1`, exposes the stream only after authenticated acceptance, and -leaves product gateway dispatch to an explicit application handler; Windows -also needs its Credential Manager and transport adapters. +granting authority. The Apple transport now includes a bounded RVT1 client, +a scoped gateway server, and a backend adapter that forwards typed RPC calls +to the existing local client while filtering state access, events, and state +changes per paired device. Product code still must connect these components to +the RVD1 session host, companion coordinator/confirmation UI, per-device scope +policy, and event fan-out; Windows also needs its Credential Manager and +transport adapters. This design follows the core idea in [DSLs for Safe iOS/watchOS Communication](https://defn.io/2025/02/16/type-safe-watchos-communication/): encode the request/response relationship in types and centralize dispatch instead of spreading dictionaries, route strings, and casts throughout both apps. Rivet generates that relationship from the shared Racket schema while keeping the exported device surface explicit. diff --git a/docs/network-device-channel.md b/docs/network-device-channel.md index 55c000a..79d7568 100644 --- a/docs/network-device-channel.md +++ b/docs/network-device-channel.md @@ -3,14 +3,16 @@ Status: active design for [issue #85](https://github.com/turinglambdaai/rivet/issues/85). The transport-independent `RivetDeviceGateway` policy/session core, pairing/ authentication lifecycle, and byte-exact session preface are implemented and -tested in memory. The Apple pairing-credential Keychain adapter, bounded -Network.framework TLS stream adapter, per-install identity provider, and -disabled-by-default macOS listener with bounded Bonjour advertising, and the -single-connection macOS preface host are also implemented. The explicit iOS -Bonjour browser accepts only the privacy-bounded canonical advertisement and -does not connect or authorize a peer. No iOS client-session coordinator, -product gateway binding, pairing UI, product enablement, or generated mobile -application is shipped yet. +tested in memory. The Apple Keychain credential store, bounded +Network.framework TLS stream, per-install identity, disabled-by-default macOS +listener and single-connection RVD1 preface host are implemented. The explicit +iOS Bonjour browser accepts only the canonical privacy-bounded advertisement. +The Apple transport now also has a bounded RVT1 client, per-device gateway +server, and backend adapter: generated APIs can run over the authenticated +network stream, while RPCs, state reads/writes, events, and state-change events +are filtered through the paired device's grants. Product code must still wire +the RVD1 host, scope policy, backend client, event fan-out, app UI, and signed +mobile application; framework code alone does not constitute a shipped app. ### iPhone, iPad and Apple Watch companion boundaries diff --git a/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift b/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift index e62b553..adab6b6 100644 --- a/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift +++ b/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift @@ -167,8 +167,8 @@ public enum RivetEvent: Sendable { } public struct RivetAPI: Sendable { - public let client: RivetClient - public init(client: RivetClient) { self.client = client } + public let client: any RivetRPCClient + public init(client: any RivetRPCClient) { self.client = client } public func create_task(title: String, notes: String) async throws -> RivetTypes.BoardTask { let rivetResult = try await client.call("create-task", arguments: [encode_String(title), encode_String(notes)]) diff --git a/platform/macos/Package.swift b/platform/macos/Package.swift index 5b5a45d..43038a4 100644 --- a/platform/macos/Package.swift +++ b/platform/macos/Package.swift @@ -57,7 +57,7 @@ let package = Package( ), .target( name: "RivetDeviceAppleTransport", - dependencies: ["RivetDevice", "RivetDeviceSession"], + dependencies: ["RivetDevice", "RivetDeviceSession", "RivetRuntime"], path: "Sources/RivetDeviceAppleTransport", linkerSettings: [ .linkedFramework("CryptoKit"), @@ -129,7 +129,12 @@ let package = Package( ), .testTarget( name: "RivetDeviceAppleTransportTests", - dependencies: ["RivetDeviceAppleTransport", "RivetDeviceSession"], + dependencies: [ + "RivetDeviceAppleTransport", + "RivetDeviceSession", + "RivetDevice", + "RivetRuntime" + ], path: "Tests/RivetDeviceAppleTransportTests" ), .testTarget( diff --git a/platform/macos/Sources/RivetDevice/DeviceGateway.swift b/platform/macos/Sources/RivetDevice/DeviceGateway.swift index 3311d1c..5ebb2b3 100644 --- a/platform/macos/Sources/RivetDevice/DeviceGateway.swift +++ b/platform/macos/Sources/RivetDevice/DeviceGateway.swift @@ -40,8 +40,10 @@ public struct RivetDeviceGatewayHostCall: Sendable { /// authorization and remote-to-host request ID translation; the adapter owns /// the actual host client and its cancellation primitive. public protocol RivetDeviceGatewayHost: Sendable { - /// Returned IDs must be nonzero and unique across simultaneously pending - /// calls. Cancellation must affect exactly the call that owns the ID. + /// Payload is an encoded RivetValue list containing the RPC arguments. + /// The adapter decodes it for the embedded backend and encodes the result + /// as a RivetValue. Returned IDs must be nonzero and unique across + /// simultaneously pending calls; cancellation must affect exactly that ID. func startRPC(name: String, payload: Data) async throws -> RivetDeviceGatewayHostCall func cancelRPC(id: UInt64) async } @@ -167,16 +169,67 @@ public actor RivetDeviceGatewaySession { remoteRequestID: UInt64, name: String, payload: Data + ) async throws -> Data { + // State routes must carry an explicit state name through the guarded + // API below; treating them as ordinary RPCs would let one granted + // state capability unlock every state of the same direction. + guard name != "$state/get", name != "$state/set" else { + throw RivetDeviceGatewayError.forbidden + } + return try await invokeRPCInternal( + remoteRequestID: remoteRequestID, + name: name, + payload: payload, + isStateRoute: false + ) + } + + package func invokeStateRPC( + remoteRequestID: UInt64, + name: String, + stateName: String, + payload: Data + ) async throws -> Data { + switch name { + case "$state/get": + try requireStateRead(stateName) + case "$state/set": + try requireStateWrite(stateName) + default: + throw RivetDeviceGatewayError.forbidden + } + return try await invokeRPCInternal( + remoteRequestID: remoteRequestID, + name: name, + payload: payload, + isStateRoute: true + ) + } + + private func invokeRPCInternal( + remoteRequestID: UInt64, + name: String, + payload: Data, + isStateRoute: Bool ) async throws -> Data { guard !closed else { throw RivetDeviceGatewayError.sessionClosed } + guard !Task.isCancelled else { + throw RivetDeviceGatewayError.cancelled + } guard remoteRequestID != 0 else { throw RivetDeviceGatewayError.invalidRequestID } - guard grants.rpcs.contains(name) else { + if isStateRoute { + guard name == "$state/get" || name == "$state/set" else { + throw RivetDeviceGatewayError.forbidden + } + } else { // Unknown and ungranted names intentionally have the same result. - throw RivetDeviceGatewayError.forbidden + guard grants.rpcs.contains(name) else { + throw RivetDeviceGatewayError.forbidden + } } try checkMessageSize(payload) guard pending[remoteRequestID] == nil else { diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Channel.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Channel.swift new file mode 100644 index 0000000..66a16ea --- /dev/null +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Channel.swift @@ -0,0 +1,127 @@ +import Foundation +import RivetDevice +import RivetRuntime + +public protocol RivetDeviceAppleApplicationByteStream: Sendable { + func sendApplicationBytes(_ data: Data) async throws + func receiveApplicationBytes(maximumLength: Int) async throws -> Data? + func close() async +} + +extension RivetDeviceAppleConnection: RivetDeviceAppleApplicationByteStream { + public func close() async { cancel() } +} + +public enum RivetDeviceAppleRVT1ChannelError: Error, Equatable, Sendable { + case closed + case concurrentRead + case frameTooLarge(actual: Int, maximum: Int) +} + +/// Bounded RVT1 framing over a stream opened only after RVD1 authentication. +public actor RivetDeviceAppleRVT1Channel { + private static let headerSize = 18 + private let stream: any RivetDeviceAppleApplicationByteStream + private let maximumPayloadSize: Int + private var readBuffer = Data() + private var reading = false + private var closed = false + private var writeTail: Task? + private var writeToken: UUID? + + public init(connection: RivetDeviceAppleConnection, + maximumPayloadSize: Int = rivetDeviceDefaultMaximumMessageSize) { + precondition(maximumPayloadSize > 0 && + maximumPayloadSize <= rivetDeviceDefaultMaximumMessageSize) + stream = connection + self.maximumPayloadSize = maximumPayloadSize + } + + package init(stream: any RivetDeviceAppleApplicationByteStream, + maximumPayloadSize: Int = rivetDeviceDefaultMaximumMessageSize) { + precondition(maximumPayloadSize > 0 && + maximumPayloadSize <= rivetDeviceDefaultMaximumMessageSize) + self.stream = stream + self.maximumPayloadSize = maximumPayloadSize + } + + public func send(_ frame: RivetFrame) async throws { + guard !closed else { throw RivetDeviceAppleRVT1ChannelError.closed } + guard frame.payload.count <= maximumPayloadSize else { + throw RivetDeviceAppleRVT1ChannelError.frameTooLarge( + actual: frame.payload.count, maximum: maximumPayloadSize + ) + } + let data = try encodeRivetFrame(frame) + let previous = writeTail + let token = UUID() + let stream = self.stream + let task = Task { + if let previous { try await previous.value } + try await stream.sendApplicationBytes(data) + } + writeTail = task + writeToken = token + do { + try await task.value + if writeToken == token { writeTail = nil; writeToken = nil } + } catch { + if writeToken == token { writeTail = nil; writeToken = nil } + if !closed { + closed = true + await stream.close() + } + throw error + } + } + + public func receive() async throws -> RivetFrame { + guard !closed else { throw RivetDeviceAppleRVT1ChannelError.closed } + guard !reading else { throw RivetDeviceAppleRVT1ChannelError.concurrentRead } + reading = true + defer { reading = false } + do { + try await fillBuffer(to: Self.headerSize) + let length = UInt32(readBuffer[14]) + | (UInt32(readBuffer[15]) << 8) + | (UInt32(readBuffer[16]) << 16) + | (UInt32(readBuffer[17]) << 24) + guard length <= UInt32(maximumPayloadSize) else { + throw RivetDeviceAppleRVT1ChannelError.frameTooLarge( + actual: Int(length), maximum: maximumPayloadSize + ) + } + let total = Self.headerSize + Int(length) + try await fillBuffer(to: total) + let frameData = Data(readBuffer.prefix(total)) + readBuffer = Data(readBuffer.dropFirst(total)) + return try decodeRivetFrame(frameData) + } catch { + // Framing and stream errors are terminal. Keeping the channel + // open would retry the same malformed bytes on the next receive. + if !closed { + closed = true + readBuffer.removeAll(keepingCapacity: false) + await stream.close() + } + throw error + } + } + + public func close() async { + guard !closed else { return } + closed = true + await stream.close() + } + + private func fillBuffer(to count: Int) async throws { + while readBuffer.count < count { + guard let chunk = try await stream.receiveApplicationBytes( + maximumLength: count - readBuffer.count + ), !chunk.isEmpty else { + throw RivetDeviceAppleRVT1ChannelError.closed + } + readBuffer.append(chunk) + } + } +} diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift new file mode 100644 index 0000000..20ccc6a --- /dev/null +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift @@ -0,0 +1,214 @@ +import Foundation +import RivetDevice +import RivetRuntime + +public enum RivetDeviceAppleRVT1ClientError: Error, Equatable, Sendable { + case alreadyStarted + case invalidHello + case invalidResponse + case invalidState + case tooManyPendingRequests(Int) + case messageTooLarge(actual: Int, maximum: Int) + case remote(String) + case stopped +} + +private struct RivetDeviceApplePendingRequest { + var continuation: CheckedContinuation? + var cancelled = false + var requestSent = false + var cancelSent = false +} + +/// Typed RVT1 client for an authenticated Apple device connection. +public actor RivetDeviceAppleRVT1Client: RivetRPCClient { + public typealias EventHandler = @Sendable (String, RivetValue) -> Void + private let channel: RivetDeviceAppleRVT1Channel + private let maximumPendingRequests: Int + private var nextRequestID: UInt64 = 1 + private var pending: [UInt64: RivetDeviceApplePendingRequest] = [:] + private var started = false + private var stopped = false + private var readerTask: Task? + private var eventHandler: EventHandler? + + public init(connection: RivetDeviceAppleConnection, + maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests) { + precondition(maximumPendingRequests > 0) + channel = RivetDeviceAppleRVT1Channel(connection: connection) + self.maximumPendingRequests = maximumPendingRequests + } + + package init(stream: any RivetDeviceAppleApplicationByteStream, + maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests) { + precondition(maximumPendingRequests > 0) + channel = RivetDeviceAppleRVT1Channel(stream: stream) + self.maximumPendingRequests = maximumPendingRequests + } + + public func start(onEvent: EventHandler? = nil) async throws { + guard !started, !stopped else { + throw RivetDeviceAppleRVT1ClientError.alreadyStarted + } + let hello: RivetFrame + do { hello = try await channel.receive() } + catch { stopped = true; await channel.close(); throw error } + guard hello.type == .hello, hello.id == 0 else { + stopped = true + await channel.close() + throw RivetDeviceAppleRVT1ClientError.invalidHello + } + do { + guard try decodeRivetValue(hello.payload) == .list([ + .string("rivet"), .int64(Int64(rivetProtocolVersion)) + ]) else { + throw RivetDeviceAppleRVT1ClientError.invalidHello + } + } catch { + stopped = true + await channel.close() + throw RivetDeviceAppleRVT1ClientError.invalidHello + } + eventHandler = onEvent + started = true + readerTask = Task { [weak self] in await self?.readLoop() } + } + + public func call(_ name: String, arguments: [RivetValue] = []) async throws -> RivetValue { + guard started, !stopped else { throw RivetDeviceAppleRVT1ClientError.invalidState } + guard pending.count < maximumPendingRequests else { + throw RivetDeviceAppleRVT1ClientError.tooManyPendingRequests(maximumPendingRequests) + } + let payload = try encodeRivetValue(.list([.string(name)] + arguments)) + guard payload.count <= rivetDeviceDefaultMaximumMessageSize else { + throw RivetDeviceAppleRVT1ClientError.messageTooLarge( + actual: payload.count, maximum: rivetDeviceDefaultMaximumMessageSize + ) + } + let id = allocateRequestID() + return try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { + (continuation: CheckedContinuation) in + pending[id] = RivetDeviceApplePendingRequest(continuation: continuation) + Task { await self.sendRequest(id: id, payload: payload) } + } + } onCancel: { + Task { await self.cancelRequest(id: id) } + } + } + + public func stop() async { + guard !stopped else { return } + stopped = true + readerTask?.cancel() + readerTask = nil + failAll(RivetDeviceAppleRVT1ClientError.stopped) + if started { try? await channel.send(RivetFrame(type: .shutdown, id: 0)) } + await channel.close() + } + + private func allocateRequestID() -> UInt64 { + for _ in 0...pending.count { + let candidate = nextRequestID + nextRequestID = nextRequestID == UInt64.max ? 1 : nextRequestID + 1 + if candidate != 0, pending[candidate] == nil { return candidate } + } + preconditionFailure("Rivet request ID allocator invariant violated") + } + + private func sendRequest(id: UInt64, payload: Data) async { + guard let entry = pending[id] else { return } + if entry.cancelled { pending.removeValue(forKey: id); return } + do { + try await channel.send(RivetFrame(type: .request, id: id, payload: payload)) + guard var current = pending[id] else { return } + current.requestSent = true + if current.cancelled && !current.cancelSent { + current.cancelSent = true + pending.removeValue(forKey: id) + Task { + do { try await channel.send(RivetFrame(type: .cancel, id: id)) } + catch { await self.failConnection(error) } + } + } else { pending[id] = current } + } catch { + stopped = true + failAll(error) + await channel.close() + } + } + + private func cancelRequest(id: UInt64) async { + guard var entry = pending[id] else { return } + entry.cancelled = true + entry.continuation?.resume(throwing: CancellationError()) + entry.continuation = nil + if entry.requestSent && !entry.cancelSent { + entry.cancelSent = true + pending.removeValue(forKey: id) + Task { + do { try await channel.send(RivetFrame(type: .cancel, id: id)) } + catch { await self.failConnection(error) } + } + } else { pending[id] = entry } + } + + private func readLoop() async { + do { + while !Task.isCancelled && !stopped { + let frame = try await channel.receive() + switch frame.type { + case .response: + guard frame.id != 0 else { + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + let value = try decodeRivetValue(frame.payload) + guard let request = pending.removeValue(forKey: frame.id) else { continue } + request.continuation?.resume(returning: value) + case .error: + guard frame.id != 0 else { + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + guard case .string(let message) = try decodeRivetValue(frame.payload) else { + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + guard let request = pending.removeValue(forKey: frame.id) else { continue } + request.continuation?.resume( + throwing: RivetDeviceAppleRVT1ClientError.remote(message) + ) + case .event: + guard frame.id == 0, + case .list(let fields) = try decodeRivetValue(frame.payload), + fields.count == 2, + case .string(let name) = fields[0] else { + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + eventHandler?(name, fields[1]) + case .shutdown: + guard frame.id == 0, frame.payload.isEmpty else { + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + throw RivetDeviceAppleRVT1ClientError.stopped + default: + throw RivetDeviceAppleRVT1ClientError.invalidResponse + } + } + } catch { + stopped = true + failAll(error) + await channel.close() + } + } + + private func failConnection(_ error: any Error) async { + stopped = true + failAll(error) + await channel.close() + } + + private func failAll(_ error: any Error) { + let requests = Array(pending.values) + pending.removeAll(keepingCapacity: true) + for request in requests { request.continuation?.resume(throwing: error) } + } +} diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift new file mode 100644 index 0000000..d3b4af8 --- /dev/null +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -0,0 +1,275 @@ +import Foundation +import RivetDevice +import RivetRuntime + +public enum RivetDeviceAppleRVT1GatewayServerError: Error, Equatable, Sendable { + case alreadyStarted + case notRunning + case invalidFrame + case invalidRequest + case invalidRequestID + case duplicateRequestID(UInt64) +} + +/// Terminates the post-RVD1 RVT1 stream and applies per-pairing gateway +/// permissions before dispatching any RPC to the embedded backend. +public actor RivetDeviceAppleRVT1GatewayServer { + private enum Route: Sendable { + case rpc + case stateRead(String) + case stateWrite(String) + } + + private let channel: RivetDeviceAppleRVT1Channel + private let gateway: RivetDeviceGateway + private let session: RivetDeviceGatewaySession + private var started = false + private var ready = false + private var stopping = false + private var closed = false + private var activeIDs = Set() + private var cancelledIDs = Set() + private var requestTasks: [UInt64: Task] = [:] + + public init( + connection: RivetDeviceAppleConnection, + gateway: RivetDeviceGateway, + session: RivetDeviceGatewaySession + ) { + channel = RivetDeviceAppleRVT1Channel(connection: connection) + self.gateway = gateway + self.session = session + } + + package init( + stream: any RivetDeviceAppleApplicationByteStream, + gateway: RivetDeviceGateway, + session: RivetDeviceGatewaySession + ) { + channel = RivetDeviceAppleRVT1Channel(stream: stream) + self.gateway = gateway + self.session = session + } + + /// Runs until the client sends Shutdown or the authenticated stream closes. + /// The gateway session is always revoked when this method exits. + public func run() async throws { + guard !started, !closed else { + throw RivetDeviceAppleRVT1GatewayServerError.alreadyStarted + } + started = true + do { + let hello = try encodeRivetValue(.list([ + .string("rivet"), + .int64(Int64(rivetProtocolVersion)) + ])) + try await channel.send(RivetFrame(type: .hello, id: 0, payload: hello)) + ready = true + while !stopping { + let frame = try await channel.receive() + switch frame.type { + case .request: + try await acceptRequest(frame) + case .cancel: + guard frame.id != 0, frame.payload.isEmpty else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + await cancelRequest(id: frame.id) + case .shutdown: + guard frame.id == 0, frame.payload.isEmpty else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + stopping = true + default: + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + } + // A peer can shut down while an RPC is waiting on an unresponsive + // backend. Revoke the session and cancel outstanding work instead + // of waiting for every response before the connection can close. + await close() + } catch { + await close() + throw error + } + } + + /// Sends an event only if this paired device has the named event grant. + @discardableResult + public func sendEvent(name: String, value: RivetValue) async throws -> Bool { + guard ready, !stopping, !closed else { + throw RivetDeviceAppleRVT1GatewayServerError.notRunning + } + let payload = try encodeRivetValue(.list([.string(name), value])) + guard let filtered = try await session.filterEvent(name: name, payload: payload) else { + return false + } + guard ready, !stopping, !closed else { + throw RivetDeviceAppleRVT1GatewayServerError.notRunning + } + try await channel.send(RivetFrame(type: .event, id: 0, payload: filtered)) + return true + } + + /// State changes use the standard $state event and are filtered against + /// the paired device's read grants. + @discardableResult + public func sendStateChange(name: String, value: RivetValue) async throws -> Bool { + guard ready, !stopping, !closed else { + throw RivetDeviceAppleRVT1GatewayServerError.notRunning + } + let payload = try encodeRivetValue(.list([ + .string("$state"), + .list([.string(name), value]) + ])) + guard let filtered = try await session.filterStateChange( + name: name, + payload: payload + ) else { + return false + } + guard ready, !stopping, !closed else { + throw RivetDeviceAppleRVT1GatewayServerError.notRunning + } + try await channel.send(RivetFrame(type: .event, id: 0, payload: filtered)) + return true + } + + public func close() async { + guard !closed else { return } + closed = true + ready = false + stopping = true + let ids = Array(activeIDs) + for id in ids { + cancelledIDs.insert(id) + await session.cancel(remoteRequestID: id) + } + let tasks = Array(requestTasks.values) + for task in tasks { + task.cancel() + } + await channel.close() + await gateway.closeSession(id: session.id) + } + + private func acceptRequest(_ frame: RivetFrame) async throws { + guard frame.id != 0 else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidRequestID + } + guard activeIDs.insert(frame.id).inserted else { + throw RivetDeviceAppleRVT1GatewayServerError.duplicateRequestID(frame.id) + } + do { + guard case .list(let fields) = try decodeRivetValue(frame.payload), + let first = fields.first, + case .string(let name) = first else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidRequest + } + let arguments = Array(fields.dropFirst()) + let route: Route + switch name { + case "$state/get": + guard arguments.count == 1, + case .string(let stateName) = arguments[0] else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidRequest + } + route = .stateRead(stateName) + case "$state/set": + guard arguments.count == 2, + case .string(let stateName) = arguments[0] else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidRequest + } + route = .stateWrite(stateName) + default: + route = .rpc + } + let payload = try encodeRivetValue(.list(arguments)) + let session = self.session + let task = Task { [weak self] in + do { + let response: Data + switch route { + case .rpc: + response = try await session.invokeRPC( + remoteRequestID: frame.id, + name: name, + payload: payload + ) + case .stateRead(let stateName), .stateWrite(let stateName): + response = try await session.invokeStateRPC( + remoteRequestID: frame.id, + name: name, + stateName: stateName, + payload: payload + ) + } + await self?.completeRequest(id: frame.id, response: response) + } catch { + await self?.completeRequest(id: frame.id, error: error) + } + } + requestTasks[frame.id] = task + } catch { + activeIDs.remove(frame.id) + try await sendError(id: frame.id, message: safeMessage(for: error)) + } + } + + private func cancelRequest(id: UInt64) async { + guard id != 0, activeIDs.contains(id) else { return } + cancelledIDs.insert(id) + // Cancel the dispatch task as well as forwarding cancellation to the + // session. If dispatch has not entered the gateway yet, its + // Task.isCancelled guard prevents a cancelled RPC from starting. + requestTasks[id]?.cancel() + await session.cancel(remoteRequestID: id) + } + + private func completeRequest( + id: UInt64, + response: Data? = nil, + error: (any Error)? = nil + ) async { + guard activeIDs.contains(id) else { return } + if cancelledIDs.remove(id) != nil || closed { + activeIDs.remove(id) + requestTasks.removeValue(forKey: id) + return + } + activeIDs.remove(id) + requestTasks.removeValue(forKey: id) + do { + if let response { + try await channel.send(RivetFrame(type: .response, id: id, payload: response)) + } else { + try await sendError(id: id, message: safeMessage(for: error)) + } + } catch { + await close() + } + } + + private func sendError(id: UInt64, message: String) async throws { + let payload = try encodeRivetValue(.string(message)) + try await channel.send(RivetFrame(type: .error, id: id, payload: payload)) + } + + private func safeMessage(for error: (any Error)?) -> String { + guard let error else { return "request failed" } + if let gatewayError = error as? RivetDeviceGatewayError { + switch gatewayError { + case .forbidden: + return "operation not permitted" + case .cancelled: + return "request cancelled" + default: + return "request failed" + } + } + if error is RivetDeviceAppleRVT1GatewayServerError { + return "invalid request" + } + return "request failed" + } +} diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift new file mode 100644 index 0000000..ef94cd0 --- /dev/null +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift @@ -0,0 +1,66 @@ +import Foundation +import RivetDevice +import RivetRuntime + +public enum RivetDeviceRVT1BackendHostError: Error, Equatable, Sendable { + case invalidArguments + case requestIDExhausted +} + +/// Adapts the embedded backend client to the gateway's bounded host-call +/// contract. Payloads are encoded RivetValue argument lists, not JSON. +public actor RivetDeviceRVT1BackendHost: RivetDeviceGatewayHost { + private let client: any RivetRPCClient + private var nextID: UInt64 = 1 + private var work: [UInt64: Task] = [:] + + public init(client: any RivetRPCClient) { + self.client = client + } + + public func startRPC( + name: String, + payload: Data + ) async throws -> RivetDeviceGatewayHostCall { + guard case .list(let arguments) = try decodeRivetValue(payload) else { + throw RivetDeviceRVT1BackendHostError.invalidArguments + } + let id = try allocateID() + let client = self.client + let workTask = Task { + let result = try await client.call(name, arguments: arguments) + return try encodeRivetValue(result) + } + work[id] = workTask + let responseTask = Task { [weak self] in + do { + let data = try await workTask.value + await self?.finish(id: id) + return data + } catch { + await self?.finish(id: id) + throw error + } + } + return RivetDeviceGatewayHostCall(id: id, response: responseTask) + } + + public func cancelRPC(id: UInt64) async { + work[id]?.cancel() + } + + private func allocateID() throws -> UInt64 { + for _ in 0...work.count { + let candidate = nextID + nextID = nextID == UInt64.max ? 1 : nextID + 1 + if candidate != 0, work[candidate] == nil { + return candidate + } + } + throw RivetDeviceRVT1BackendHostError.requestIDExhausted + } + + private func finish(id: UInt64) { + work.removeValue(forKey: id) + } +} diff --git a/platform/macos/Sources/RivetRuntime/RivetRPCClient.swift b/platform/macos/Sources/RivetRuntime/RivetRPCClient.swift new file mode 100644 index 0000000..7dba694 --- /dev/null +++ b/platform/macos/Sources/RivetRuntime/RivetRPCClient.swift @@ -0,0 +1,21 @@ +import Foundation + +/// The minimal typed RPC surface consumed by generated \`RivetAPI\` code. +/// In-process clients and authenticated network clients can share the same +/// generated API without pretending to have the same lifecycle or transport. +public protocol RivetRPCClient: Sendable { + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue +} + +public extension RivetRPCClient { + func getState(_ name: String) async throws -> RivetValue { + try await call("$state/get", arguments: [.string(name)]) + } + + @discardableResult + func setState(_ name: String, value: RivetValue) async throws -> RivetValue { + try await call("$state/set", arguments: [.string(name), value]) + } +} + +extension RivetClient: RivetRPCClient {} diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift new file mode 100644 index 0000000..062ad68 --- /dev/null +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -0,0 +1,348 @@ +import Foundation +import RivetDevice +import RivetRuntime +import Testing +@testable import RivetDeviceAppleTransport + +private actor MemoryByteStream: RivetDeviceAppleApplicationByteStream { + private struct Waiter { + let maximumLength: Int + let continuation: CheckedContinuation + } + + private var incoming: Data + private var sent: [Data] = [] + private var waiters: [Waiter] = [] + private var closed = false + + init(incoming: Data = Data()) { self.incoming = incoming } + + func sendApplicationBytes(_ data: Data) async throws { + guard !closed else { throw RivetDeviceAppleRVT1ChannelError.closed } + sent.append(data) + } + + func receiveApplicationBytes(maximumLength: Int) async throws -> Data? { + if !incoming.isEmpty { return take(maximumLength) } + if closed { return nil } + return try await withCheckedThrowingContinuation { + (continuation: CheckedContinuation) in + waiters.append(Waiter(maximumLength: maximumLength, continuation: continuation)) + } + } + + func close() async { + closed = true + resumeWaiters() + } + + func enqueue(_ data: Data) { + guard !closed else { return } + incoming.append(data) + resumeWaiters() + } + + func sentCount() -> Int { sent.count } + func isClosed() -> Bool { closed } + + func sentFrames() throws -> [RivetFrame] { + try sent.map { try decodeRivetFrame($0) } + } + + private func take(_ maximumLength: Int) -> Data { + let count = min(maximumLength, incoming.count) + let result = Data(incoming.prefix(count)) + incoming = Data(incoming.dropFirst(count)) + return result + } + + private func resumeWaiters() { + while !waiters.isEmpty && (!incoming.isEmpty || closed) { + let waiter = waiters.removeFirst() + if incoming.isEmpty { + waiter.continuation.resume(returning: nil) + } else { + waiter.continuation.resume(returning: take(waiter.maximumLength)) + } + } + } +} + +private actor SlowGatewayHost: RivetDeviceGatewayHost { + private var started = false + private var cancellations = 0 + private var responseTask: Task? + + func startRPC(name: String, payload: Data) async throws -> RivetDeviceGatewayHostCall { + started = true + let task = Task { + try await Task.sleep(nanoseconds: 30_000_000_000) + return Data([0x01]) + } + responseTask = task + return RivetDeviceGatewayHostCall(id: 1, response: task) + } + + func cancelRPC(id: UInt64) async { + cancellations += 1 + responseTask?.cancel() + } + + func didStart() -> Bool { started } + func cancellationCount() -> Int { cancellations } +} + +private actor EchoRPCClient: RivetRPCClient { + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue { + .list([.string(name)] + arguments) + } +} + +private actor TestGatewayHost: RivetDeviceGatewayHost { + private var callNames: [String] = [] + + func startRPC(name: String, payload: Data) async throws -> RivetDeviceGatewayHostCall { + callNames.append(name) + let response = try encodeRivetValue(.string("host-ok")) + let task = Task { response } + return RivetDeviceGatewayHostCall(id: UInt64(callNames.count), response: task) + } + + func cancelRPC(id: UInt64) async {} + func names() -> [String] { callNames } +} + +private func encodedFrame(_ frame: RivetFrame) throws -> Data { + try encodeRivetFrame(frame) +} + +private func helloFrame() throws -> Data { + try encodedFrame(RivetFrame( + type: .hello, + id: 0, + payload: try encodeRivetValue(.list([ + .string("rivet"), + .int64(Int64(rivetProtocolVersion)) + ])) + )) +} + +@Test func appleRVT1ChannelClosesAfterOversizedFrame() async throws { + let frame = try encodedFrame(RivetFrame(type: .event, id: 0, payload: Data([1, 2]))) + let stream = MemoryByteStream(incoming: frame) + let channel = RivetDeviceAppleRVT1Channel(stream: stream, maximumPayloadSize: 1) + + await #expect(throws: RivetDeviceAppleRVT1ChannelError.frameTooLarge( + actual: 2, maximum: 1 + )) { + try await channel.receive() + } + #expect(await stream.isClosed()) +} + +@Test func appleRVT1ClientPerformsHelloAndCorrelatesTypedResponses() async throws { + let stream = MemoryByteStream(incoming: try helloFrame()) + let client = RivetDeviceAppleRVT1Client(stream: stream) + try await client.start() + let call = Task { try await client.call("player.seek", arguments: [.int64(42)]) } + for _ in 0..<2_000 { + if await stream.sentCount() == 1 { break } + await Task.yield() + } + let sent = try await stream.sentFrames() + let request = try #require(sent.first) + #expect(request.type == .request) + #expect(try decodeRivetValue(request.payload) == .list([ + .string("player.seek"), .int64(42) + ])) + await stream.enqueue(try encodedFrame(RivetFrame( + type: .response, + id: request.id, + payload: try encodeRivetValue(.string("seeked")) + ))) + #expect(try await call.value == .string("seeked")) + await client.stop() +} + +@Test func appleRVT1ClientCancellationSendsMatchingCancelFrame() async throws { + let stream = MemoryByteStream(incoming: try helloFrame()) + let client = RivetDeviceAppleRVT1Client(stream: stream) + try await client.start() + let call = Task { + try await client.call("player.wait", arguments: [.int64(5)]) + } + for _ in 0..<2_000 { + if await stream.sentCount() >= 1 { break } + await Task.yield() + } + let sentBeforeCancel = try await stream.sentFrames() + let request = try #require(sentBeforeCancel.first) + #expect(request.type == .request) + + call.cancel() + await #expect(throws: CancellationError.self) { + try await call.value + } + for _ in 0..<2_000 { + if await stream.sentCount() >= 2 { break } + await Task.yield() + } + let sent = try await stream.sentFrames() + #expect(sent.count == 2) + #expect(sent[1].type == .cancel) + #expect(sent[1].id == request.id) + await client.stop() +} + +@Test func appleRVT1GatewayEnforcesRPCAndStateGrants() async throws { + let host = TestGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope( + name: "companion", + rpcs: ["player.play"], + readableStates: ["volume"] + )] + ) + var incoming = Data() + let frames = [ + RivetFrame(type: .request, id: 7, + payload: try encodeRivetValue(.list([.string("player.play")]))), + RivetFrame(type: .request, id: 8, + payload: try encodeRivetValue(.list([.string("$state/get"), .string("volume")]))), + RivetFrame(type: .request, id: 9, + payload: try encodeRivetValue(.list([.string("$state/get"), .string("private.state")]))), + RivetFrame(type: .shutdown, id: 0) + ] + for frame in frames { incoming.append(try encodedFrame(frame)) } + let stream = MemoryByteStream(incoming: incoming) + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + try await server.run() + + let sent = try await stream.sentFrames() + #expect(sent.first?.type == .hello) + let byID = Dictionary(uniqueKeysWithValues: sent.filter { + $0.type == .response || $0.type == .error + }.map { ($0.id, $0) }) + #expect(byID[7]?.type == .response) + #expect(byID[8]?.type == .response) + #expect(byID[9]?.type == .error) + let denied = try #require(byID[9]) + #expect(try decodeRivetValue(denied.payload) == .string("operation not permitted")) + #expect(Set(await host.names()) == Set(["player.play", "$state/get"])) + #expect(await gateway.sessionCount() == 0) +} + +@Test func appleRVT1GatewayCancelForwardsCancellationAndSuppressesLateResponse() async throws { + let host = SlowGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + let request = try encodedFrame(RivetFrame( + type: .request, + id: 56, + payload: try encodeRivetValue(.list([.string("player.play")])) + )) + let stream = MemoryByteStream(incoming: request) + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + + for _ in 0..<2_000 { + if await host.didStart() { break } + await Task.yield() + } + #expect(await host.didStart()) + await stream.enqueue(try encodedFrame(RivetFrame(type: .cancel, id: 56))) + for _ in 0..<2_000 { + if await host.cancellationCount() >= 1 { break } + await Task.yield() + } + #expect(await host.cancellationCount() >= 1) + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value + + let sent = try await stream.sentFrames() + #expect(sent.map(\.type) == [.hello]) + #expect(await gateway.sessionCount() == 0) +} + +@Test func appleRVT1GatewayShutdownCancelsInFlightRPCAndRevokesSession() async throws { + let host = SlowGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + let request = try encodedFrame(RivetFrame( + type: .request, + id: 55, + payload: try encodeRivetValue(.list([.string("player.play")])) + )) + let stream = MemoryByteStream(incoming: request) + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + + for _ in 0..<2_000 { + if await host.didStart() { break } + await Task.yield() + } + #expect(await host.didStart()) + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value + for _ in 0..<2_000 { + if await host.cancellationCount() >= 1 { break } + await Task.yield() + } + + #expect(await host.cancellationCount() >= 1) + #expect(await gateway.sessionCount() == 0) +} + +@Test func appleRVT1GatewayFiltersEventsAndStateChangesPerDevice() async throws { + let host = TestGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope( + name: "companion", + events: ["progress"], + readableStates: ["volume"] + )] + ) + let stream = MemoryByteStream() + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + for _ in 0..<2_000 { + if await stream.sentCount() >= 1 { break } + await Task.yield() + } + #expect(try await server.sendEvent(name: "progress", value: .int64(3))) + #expect(!(try await server.sendEvent(name: "private.event", value: .int64(3)))) + #expect(try await server.sendStateChange(name: "volume", value: .int64(8))) + #expect(!(try await server.sendStateChange(name: "private.state", value: .int64(8)))) + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value + let sent = try await stream.sentFrames() + #expect(sent.map(\.type) == [.hello, .event, .event]) +} + +@Test func rvt1BackendHostEncodesArgumentsAndResults() async throws { + let host = RivetDeviceRVT1BackendHost(client: EchoRPCClient()) + let payload = try encodeRivetValue(.list([.string("chapter"), .int64(4)])) + let call = try await host.startRPC(name: "player.seek", payload: payload) + let response = try await call.response.value + #expect(try decodeRivetValue(response) == .list([ + .string("player.seek"), .string("chapter"), .int64(4) + ])) +} diff --git a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift index 4cff310..199cce9 100644 --- a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift @@ -101,6 +101,42 @@ private let pairedScope = RivetDeviceScope( #expect(starts.map(\.name) == ["tasks.list"]) } +@Test func stateRoutesRequireTheExactStateCapability() async throws { + let host = TestGatewayHost() + let session = try RivetDeviceGatewaySession( + pairedDeviceID: UUID(), + scopes: [pairedScope], + host: host + ) + let payload = Data("encoded state args".utf8) + + await #expect(throws: RivetDeviceGatewayError.forbidden) { + try await session.invokeRPC( + remoteRequestID: 20, + name: "$state/get", + payload: payload + ) + } + await #expect(throws: RivetDeviceGatewayError.forbidden) { + try await session.invokeStateRPC( + remoteRequestID: 21, + name: "$state/get", + stateName: "private.state", + payload: payload + ) + } + + let response = try await session.invokeStateRPC( + remoteRequestID: 22, + name: "$state/get", + stateName: "sync.status", + payload: payload + ) + #expect(response == Data([0x42])) + let starts = await host.starts + #expect(starts.map(\.name) == ["$state/get"]) +} + @Test func remoteRequestIDsMapToHostIDsAndCancellationStaysInSession() async throws { let host = TestGatewayHost(blocksResponses: true) let first = try RivetDeviceGatewaySession( diff --git a/platform/macos/Tests/RivetRuntimeTests/RivetRPCClientTests.swift b/platform/macos/Tests/RivetRuntimeTests/RivetRPCClientTests.swift new file mode 100644 index 0000000..2245d72 --- /dev/null +++ b/platform/macos/Tests/RivetRuntimeTests/RivetRPCClientTests.swift @@ -0,0 +1,32 @@ +import Foundation +import Testing +@testable import RivetRuntime + +private actor RecordingRPCClient: RivetRPCClient { + private var recorded: [(String, [RivetValue])] = [] + + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue { + recorded.append((name, arguments)) + return .string(name) + } + + func snapshot() -> [(String, [RivetValue])] { + recorded + } +} + +@Test func rpcClientStateHelpersUseReservedRoutes() async throws { + let client = RecordingRPCClient() + #expect(try await client.getState("volume") == .string("$state/get")) + #expect( + try await client.setState("volume", value: .int64(7)) + == .string("$state/set") + ) + + let requests = await client.snapshot() + #expect(requests.count == 2) + #expect(requests[0].0 == "$state/get") + #expect(requests[0].1 == [.string("volume")]) + #expect(requests[1].0 == "$state/set") + #expect(requests[1].1 == [.string("volume"), .int64(7)]) +} diff --git a/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift b/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift index b711b43..d3d69e4 100644 --- a/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift +++ b/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift @@ -18,9 +18,9 @@ public enum RivetGeneratedConfig { } public struct RivetAPI: Sendable { - public let client: RivetClient + public let client: any RivetRPCClient - public init(client: RivetClient) { + public init(client: any RivetRPCClient) { self.client = client } diff --git a/rivet-cli/codegen/swift.rkt b/rivet-cli/codegen/swift.rkt index 6288e86..d93cea1 100644 --- a/rivet-cli/codegen/swift.rkt +++ b/rivet-cli/codegen/swift.rkt @@ -309,7 +309,7 @@ (apply string-append (map swift-decoder types)) "\n" (generate-swift-events events) - "public struct RivetAPI: Sendable {\n public let client: RivetClient\n public init(client: RivetClient) { self.client = client }\n\n" + "public struct RivetAPI: Sendable {\n public let client: any RivetRPCClient\n public init(client: any RivetRPCClient) { self.client = client }\n\n" (apply string-append (map swift-rpc-method rpcs)) (if (null? states) "" "\n // Shared state\n") (apply string-append (map swift-state-methods states)) diff --git a/rivet/tests/codegen.rkt b/rivet/tests/codegen.rkt index da376c5..a0a8804 100644 --- a/rivet/tests/codegen.rkt +++ b/rivet/tests/codegen.rkt @@ -219,6 +219,8 @@ RKT ".rivet" "generated" "kotlin" "dev" "rivet" "generated" "GeneratedBackend.kt"))) + (check-true (regexp-match? #rx"public let client: any RivetRPCClient" swift)) + (check-true (regexp-match? #rx"public init\\(client: any RivetRPCClient\\)" swift)) (check-true (regexp-match? #rx"public enum RivetTypes" swift)) (check-true (regexp-match? #rx" public struct User: Sendable" swift)) (check-true (regexp-match? #rx"displayName = \"demo\"" swift)) From 8db12448aab847a22efd92ace9963d1139ccc74e Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 14:48:53 +0800 Subject: [PATCH 02/12] test(apple): drain gateway replies before graceful shutdown --- .../AppleRVT1GatewayTests.swift | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 062ad68..4c2f376 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -213,14 +213,20 @@ private func helloFrame() throws -> Data { payload: try encodeRivetValue(.list([.string("$state/get"), .string("volume")]))), RivetFrame(type: .request, id: 9, payload: try encodeRivetValue(.list([.string("$state/get"), .string("private.state")]))), - RivetFrame(type: .shutdown, id: 0) ] for frame in frames { incoming.append(try encodedFrame(frame)) } let stream = MemoryByteStream(incoming: incoming) let server = RivetDeviceAppleRVT1GatewayServer( stream: stream, gateway: gateway, session: session ) - try await server.run() + let running = Task { try await server.run() } + for _ in 0..<2_000 { + if await stream.sentCount() >= 4 { break } + await Task.yield() + } + #expect(await stream.sentCount() >= 4) + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value let sent = try await stream.sentFrames() #expect(sent.first?.type == .hello) From abaed0358902dd554e516309ca007477970cae7d Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 14:51:56 +0800 Subject: [PATCH 03/12] fix(apple): bound concurrent RVT1 gateway requests --- .../RivetDeviceAppleRVT1GatewayServer.swift | 7 ++- .../AppleRVT1GatewayTests.swift | 54 ++++++++++++++++--- 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift index d3b4af8..3831a09 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -157,9 +157,14 @@ public actor RivetDeviceAppleRVT1GatewayServer { guard frame.id != 0 else { throw RivetDeviceAppleRVT1GatewayServerError.invalidRequestID } - guard activeIDs.insert(frame.id).inserted else { + guard !activeIDs.contains(frame.id) else { throw RivetDeviceAppleRVT1GatewayServerError.duplicateRequestID(frame.id) } + guard activeIDs.count < rivetDeviceDefaultMaximumPendingRequests else { + try await sendError(id: frame.id, message: "too many pending requests") + return + } + activeIDs.insert(frame.id) do { guard case .list(let fields) = try decodeRivetValue(frame.payload), let first = fields.first, diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 4c2f376..2f13561 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -69,26 +69,30 @@ private actor MemoryByteStream: RivetDeviceAppleApplicationByteStream { } private actor SlowGatewayHost: RivetDeviceGatewayHost { - private var started = false + private var starts = 0 private var cancellations = 0 - private var responseTask: Task? + private var nextID: UInt64 = 1 + private var responseTasks: [UInt64: Task] = [:] func startRPC(name: String, payload: Data) async throws -> RivetDeviceGatewayHostCall { - started = true + starts += 1 + let id = nextID + nextID += 1 let task = Task { try await Task.sleep(nanoseconds: 30_000_000_000) return Data([0x01]) } - responseTask = task - return RivetDeviceGatewayHostCall(id: 1, response: task) + responseTasks[id] = task + return RivetDeviceGatewayHostCall(id: id, response: task) } func cancelRPC(id: UInt64) async { cancellations += 1 - responseTask?.cancel() + responseTasks[id]?.cancel() } - func didStart() -> Bool { started } + func didStart() -> Bool { starts > 0 } + func startCount() -> Int { starts } func cancellationCount() -> Int { cancellations } } @@ -242,6 +246,42 @@ private func helloFrame() throws -> Data { #expect(await gateway.sessionCount() == 0) } +@Test func appleRVT1GatewayBoundsConcurrentRequests() async throws { + let host = SlowGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + var incoming = Data() + for id in 1...65 { + incoming.append(try encodedFrame(RivetFrame( + type: .request, + id: UInt64(id), + payload: try encodeRivetValue(.list([.string("player.play")])) + ))) + } + let stream = MemoryByteStream(incoming: incoming) + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + + for _ in 0..<2_000 { + if await stream.sentCount() >= 2 { break } + await Task.yield() + } + let sent = try await stream.sentFrames() + let rejected = try #require(sent.first(where: { $0.id == 65 })) + #expect(rejected.type == .error) + #expect(try decodeRivetValue(rejected.payload) == .string("too many pending requests")) + #expect(await host.startCount() <= rivetDeviceDefaultMaximumPendingRequests) + + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value + #expect(await gateway.sessionCount() == 0) +} + @Test func appleRVT1GatewayCancelForwardsCancellationAndSuppressesLateResponse() async throws { let host = SlowGatewayHost() let gateway = RivetDeviceGateway(host: host) From 713a99adfcf938e3a473d8e1e19d354642ab08dd Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 14:59:54 +0800 Subject: [PATCH 04/12] feat(apple): keep RVT1 application handlers alive until disconnect --- CHANGELOG.md | 13 +++--- docs/network-device-channel.md | 5 ++- .../RivetDeviceAppleRVT1Client.swift | 44 ++++++++++++++++++- .../AppleRVT1GatewayTests.swift | 28 ++++++++++++ 4 files changed, 80 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 297c5c0..be51b23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,13 @@ - Add the Apple post-RVD1 RVT1 device path: generated Swift APIs can target a shared RivetRPCClient protocol; a bounded network client correlates RPCs, - forwards cancellation and delivers events; a scoped gateway server applies - RPC/state/event grants and a backend adapter calls the existing local - client. Framework tests cover request correlation, state grant isolation, - event filtering and host argument/result encoding. Product listener wiring, - scope selection, event fan-out, signed apps and physical-device validation - remain outstanding. + forwards cancellation and delivers events, and its lifecycle-aware run() + keeps companion handlers alive until disconnect. A scoped gateway applies + RPC/state/event grants and a backend adapter calls the existing local client. + Tests cover request correlation, cancellation, shutdown, request bounds, + state-grant isolation, event filtering and host argument/result encoding. + Product listener wiring, scope selection, event fan-out, signed apps and + physical-device validation remain outstanding. - Bridge Android companion TLS streams to the RVD1 client through a one-shot selected-endpoint session driver. Enforce the stored SPKI pin on diff --git a/docs/network-device-channel.md b/docs/network-device-channel.md index 79d7568..8c235f9 100644 --- a/docs/network-device-channel.md +++ b/docs/network-device-channel.md @@ -9,8 +9,9 @@ listener and single-connection RVD1 preface host are implemented. The explicit iOS Bonjour browser accepts only the canonical privacy-bounded advertisement. The Apple transport now also has a bounded RVT1 client, per-device gateway server, and backend adapter: generated APIs can run over the authenticated -network stream, while RPCs, state reads/writes, events, and state-change events -are filtered through the paired device's grants. Product code must still wire +network stream. The client offers a lifecycle-aware `run()` entry point for +companion handlers, and RPCs, state reads/writes, events, and state-change +events are filtered through the paired device's grants. Product code must still wire the RVD1 host, scope policy, backend client, event fan-out, app UI, and signed mobile application; framework code alone does not constitute a shipped app. diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift index 20ccc6a..35e4763 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift @@ -31,6 +31,8 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { private var stopped = false private var readerTask: Task? private var eventHandler: EventHandler? + private var terminated = false + private var terminationWaiters: [CheckedContinuation] = [] public init(connection: RivetDeviceAppleConnection, maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests) { @@ -74,6 +76,26 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { readerTask = Task { [weak self] in await self?.readLoop() } } + /// Runs the RVT1 session until the peer disconnects or this client stops. + /// Unlike start(), this keeps an application-handler task alive for the + /// full connection lifetime, which is useful for companion coordinators. + public func run(onEvent: EventHandler? = nil) async throws { + try await start(onEvent: onEvent) + await withTaskCancellationHandler { + await withCheckedContinuation { + (continuation: CheckedContinuation) in + if terminated { + continuation.resume() + } else { + terminationWaiters.append(continuation) + } + } + } onCancel: { + Task { await self.stop() } + } + try Task.checkCancellation() + } + public func call(_ name: String, arguments: [RivetValue] = []) async throws -> RivetValue { guard started, !stopped else { throw RivetDeviceAppleRVT1ClientError.invalidState } guard pending.count < maximumPendingRequests else { @@ -98,13 +120,22 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } public func stop() async { - guard !stopped else { return } + guard !stopped else { + if let readerTask { await readerTask.value } + return + } stopped = true - readerTask?.cancel() + let task = readerTask readerTask = nil + task?.cancel() failAll(RivetDeviceAppleRVT1ClientError.stopped) if started { try? await channel.send(RivetFrame(type: .shutdown, id: 0)) } await channel.close() + if let task { + await task.value + } else { + markTerminated() + } } private func allocateRequestID() -> UInt64 { @@ -154,6 +185,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } private func readLoop() async { + defer { markTerminated() } do { while !Task.isCancelled && !stopped { let frame = try await channel.receive() @@ -200,6 +232,14 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } } + private func markTerminated() { + guard !terminated else { return } + terminated = true + let waiters = terminationWaiters + terminationWaiters.removeAll(keepingCapacity: false) + for waiter in waiters { waiter.resume() } + } + private func failConnection(_ error: any Error) async { stopped = true failAll(error) diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 2f13561..219f368 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -44,6 +44,7 @@ private actor MemoryByteStream: RivetDeviceAppleApplicationByteStream { func sentCount() -> Int { sent.count } func isClosed() -> Bool { closed } + func waitingReaderCount() -> Int { waiters.count } func sentFrames() throws -> [RivetFrame] { try sent.map { try decodeRivetFrame($0) } @@ -96,6 +97,12 @@ private actor SlowGatewayHost: RivetDeviceGatewayHost { func cancellationCount() -> Int { cancellations } } +private actor RunCompletion { + private var returned = false + func markReturned() { returned = true } + func didReturn() -> Bool { returned } +} + private actor EchoRPCClient: RivetRPCClient { func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue { .list([.string(name)] + arguments) @@ -168,6 +175,27 @@ private func helloFrame() throws -> Data { await client.stop() } +@Test func appleRVT1ClientRunOwnsConnectionLifetimeUntilStopped() async throws { + let stream = MemoryByteStream(incoming: try helloFrame()) + let client = RivetDeviceAppleRVT1Client(stream: stream) + let completion = RunCompletion() + let running = Task { + try await client.run() + await completion.markReturned() + } + + for _ in 0..<2_000 { + if await stream.waitingReaderCount() > 0 { break } + await Task.yield() + } + #expect(await stream.waitingReaderCount() > 0) + #expect(!(await completion.didReturn())) + + await client.stop() + try await running.value + #expect(await completion.didReturn()) +} + @Test func appleRVT1ClientCancellationSendsMatchingCancelFrame() async throws { let stream = MemoryByteStream(incoming: try helloFrame()) let client = RivetDeviceAppleRVT1Client(stream: stream) From 50a10005e92b453b233c249294c5a841739e0f2d Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 15:01:41 +0800 Subject: [PATCH 05/12] fix(apple): propagate terminal RVT1 protocol failures to run callers --- .../RivetDeviceAppleRVT1Client.swift | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift index 35e4763..ee6b9b9 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift @@ -32,6 +32,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { private var readerTask: Task? private var eventHandler: EventHandler? private var terminated = false + private var terminalError: (any Error)? private var terminationWaiters: [CheckedContinuation] = [] public init(connection: RivetDeviceAppleConnection, @@ -93,6 +94,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } onCancel: { Task { await self.stop() } } + if let terminalError { throw terminalError } try Task.checkCancellation() } @@ -163,6 +165,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } } else { pending[id] = current } } catch { + if !stopped && !isNormalTermination(error) { terminalError = error } stopped = true failAll(error) await channel.close() @@ -226,6 +229,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } } } catch { + if !stopped && !isNormalTermination(error) { terminalError = error } stopped = true failAll(error) await channel.close() @@ -241,11 +245,24 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { } private func failConnection(_ error: any Error) async { + if !stopped && !isNormalTermination(error) { terminalError = error } stopped = true failAll(error) await channel.close() } + private func isNormalTermination(_ error: any Error) -> Bool { + if let clientError = error as? RivetDeviceAppleRVT1ClientError, + clientError == .stopped { + return true + } + if let channelError = error as? RivetDeviceAppleRVT1ChannelError, + channelError == .closed { + return true + } + return false + } + private func failAll(_ error: any Error) { let requests = Array(pending.values) pending.removeAll(keepingCapacity: true) From 10a3f107330c7a4f27a9acb0e2e89c5cbfb451ff Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 15:06:47 +0800 Subject: [PATCH 06/12] fix(device): reserve state-change events for state grants --- platform/macos/Sources/RivetDevice/DeviceGateway.swift | 4 +++- .../AppleRVT1GatewayTests.swift | 6 +++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/platform/macos/Sources/RivetDevice/DeviceGateway.swift b/platform/macos/Sources/RivetDevice/DeviceGateway.swift index 5ebb2b3..0c89b41 100644 --- a/platform/macos/Sources/RivetDevice/DeviceGateway.swift +++ b/platform/macos/Sources/RivetDevice/DeviceGateway.swift @@ -309,7 +309,9 @@ public actor RivetDeviceGatewaySession { guard !closed else { throw RivetDeviceGatewayError.sessionClosed } - guard grants.events.contains(name) else { + // State changes have a separate state-name authorization path; + // never let an ordinary event grant bypass readable-state filtering. + guard name != "$state", grants.events.contains(name) else { return nil } try checkMessageSize(payload) diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 219f368..6c18e58 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -388,7 +388,7 @@ private func helloFrame() throws -> Data { pairedDeviceID: UUID(), scopes: [RivetDeviceScope( name: "companion", - events: ["progress"], + events: ["progress", "$state"], readableStates: ["volume"] )] ) @@ -403,6 +403,10 @@ private func helloFrame() throws -> Data { } #expect(try await server.sendEvent(name: "progress", value: .int64(3))) #expect(!(try await server.sendEvent(name: "private.event", value: .int64(3)))) + #expect(!(try await server.sendEvent( + name: "$state", + value: .list([.string("private.state"), .int64(9)]) + ))) #expect(try await server.sendStateChange(name: "volume", value: .int64(8))) #expect(!(try await server.sendStateChange(name: "private.state", value: .int64(8)))) await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) From ae109e9eb04ca08baea7d2cc1cc0311e49d48faf Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 15:10:04 +0800 Subject: [PATCH 07/12] fix(apple): make RVT1 handshake cancellation unblock reads --- .../RivetDeviceAppleRVT1Client.swift | 39 +++++++++++++------ .../AppleRVT1GatewayTests.swift | 17 ++++++++ 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift index ee6b9b9..2e5450e 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift @@ -54,8 +54,18 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { throw RivetDeviceAppleRVT1ClientError.alreadyStarted } let hello: RivetFrame - do { hello = try await channel.receive() } - catch { stopped = true; await channel.close(); throw error } + do { + hello = try await withTaskCancellationHandler { + try await channel.receive() + } onCancel: { + Task { await self.stop() } + } + } catch { + stopped = true + await channel.close() + if Task.isCancelled { throw CancellationError() } + throw error + } guard hello.type == .hello, hello.id == 0 else { stopped = true await channel.close() @@ -81,18 +91,23 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { /// Unlike start(), this keeps an application-handler task alive for the /// full connection lifetime, which is useful for companion coordinators. public func run(onEvent: EventHandler? = nil) async throws { - try await start(onEvent: onEvent) - await withTaskCancellationHandler { - await withCheckedContinuation { - (continuation: CheckedContinuation) in - if terminated { - continuation.resume() - } else { - terminationWaiters.append(continuation) + do { + try await withTaskCancellationHandler { + try await start(onEvent: onEvent) + await withCheckedContinuation { + (continuation: CheckedContinuation) in + if terminated { + continuation.resume() + } else { + terminationWaiters.append(continuation) + } } + } onCancel: { + Task { await self.stop() } } - } onCancel: { - Task { await self.stop() } + } catch { + if Task.isCancelled { throw CancellationError() } + throw error } if let terminalError { throw terminalError } try Task.checkCancellation() diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 6c18e58..4ec60f8 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -175,6 +175,23 @@ private func helloFrame() throws -> Data { await client.stop() } +@Test func appleRVT1ClientStartCancellationClosesPendingHandshake() async throws { + let stream = MemoryByteStream() + let client = RivetDeviceAppleRVT1Client(stream: stream) + let starting = Task { try await client.start() } + + for _ in 0..<2_000 { + if await stream.waitingReaderCount() > 0 { break } + await Task.yield() + } + #expect(await stream.waitingReaderCount() > 0) + starting.cancel() + await #expect(throws: CancellationError.self) { + try await starting.value + } + #expect(await stream.isClosed()) +} + @Test func appleRVT1ClientRunOwnsConnectionLifetimeUntilStopped() async throws { let stream = MemoryByteStream(incoming: try helloFrame()) let client = RivetDeviceAppleRVT1Client(stream: stream) From 7496a00bffa80023620c40f32a83cb4ccfc4a32f Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 15:14:51 +0800 Subject: [PATCH 08/12] fix(apple): close and revoke gateway when run task is cancelled --- .../RivetDeviceAppleRVT1GatewayServer.swift | 65 +++++++++++-------- .../AppleRVT1GatewayTests.swift | 26 ++++++++ 2 files changed, 63 insertions(+), 28 deletions(-) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift index 3831a09..eb6115a 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -59,41 +59,50 @@ public actor RivetDeviceAppleRVT1GatewayServer { } started = true do { - let hello = try encodeRivetValue(.list([ - .string("rivet"), - .int64(Int64(rivetProtocolVersion)) - ])) - try await channel.send(RivetFrame(type: .hello, id: 0, payload: hello)) - ready = true - while !stopping { - let frame = try await channel.receive() - switch frame.type { - case .request: - try await acceptRequest(frame) - case .cancel: - guard frame.id != 0, frame.payload.isEmpty else { - throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame - } - await cancelRequest(id: frame.id) - case .shutdown: - guard frame.id == 0, frame.payload.isEmpty else { - throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame - } - stopping = true - default: - throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame - } + try await withTaskCancellationHandler { + try await self.runLoop() + } onCancel: { + Task { await self.close() } } - // A peer can shut down while an RPC is waiting on an unresponsive - // backend. Revoke the session and cancel outstanding work instead - // of waiting for every response before the connection can close. - await close() } catch { await close() + if Task.isCancelled { throw CancellationError() } throw error } } + private func runLoop() async throws { + let hello = try encodeRivetValue(.list([ + .string("rivet"), + .int64(Int64(rivetProtocolVersion)) + ])) + try await channel.send(RivetFrame(type: .hello, id: 0, payload: hello)) + ready = true + while !stopping { + let frame = try await channel.receive() + switch frame.type { + case .request: + try await acceptRequest(frame) + case .cancel: + guard frame.id != 0, frame.payload.isEmpty else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + await cancelRequest(id: frame.id) + case .shutdown: + guard frame.id == 0, frame.payload.isEmpty else { + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + stopping = true + default: + throw RivetDeviceAppleRVT1GatewayServerError.invalidFrame + } + } + // A peer can shut down while an RPC is waiting on an unresponsive + // backend. Revoke the session and cancel outstanding work instead + // of waiting for every response before the connection can close. + await close() + } + /// Sends an event only if this paired device has the named event grant. @discardableResult public func sendEvent(name: String, value: RivetValue) async throws -> Bool { diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 4ec60f8..9564aad 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -398,6 +398,32 @@ private func helloFrame() throws -> Data { #expect(await gateway.sessionCount() == 0) } +@Test func appleRVT1GatewayRunCancellationClosesConnectionAndRevokesSession() async throws { + let host = TestGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + let stream = MemoryByteStream() + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + + for _ in 0..<2_000 { + if await stream.waitingReaderCount() > 0 { break } + await Task.yield() + } + #expect(await stream.waitingReaderCount() > 0) + running.cancel() + await #expect(throws: CancellationError.self) { + try await running.value + } + #expect(await stream.isClosed()) + #expect(await gateway.sessionCount() == 0) +} + @Test func appleRVT1GatewayFiltersEventsAndStateChangesPerDevice() async throws { let host = TestGatewayHost() let gateway = RivetDeviceGateway(host: host) From 657765256c4747e85d174515c120c1e70daf2a42 Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 15:20:46 +0800 Subject: [PATCH 09/12] fix(apple): treat peer disconnect as normal gateway termination --- .../RivetDeviceAppleRVT1GatewayServer.swift | 4 ++++ .../AppleRVT1GatewayTests.swift | 23 +++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift index eb6115a..e5eeaa2 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -67,6 +67,10 @@ public actor RivetDeviceAppleRVT1GatewayServer { } catch { await close() if Task.isCancelled { throw CancellationError() } + if let channelError = error as? RivetDeviceAppleRVT1ChannelError, + channelError == .closed { + return + } throw error } } diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 9564aad..c7d53a2 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -398,6 +398,29 @@ private func helloFrame() throws -> Data { #expect(await gateway.sessionCount() == 0) } +@Test func appleRVT1GatewayPeerDisconnectRevokesSessionNormally() async throws { + let host = TestGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + let stream = MemoryByteStream() + let server = RivetDeviceAppleRVT1GatewayServer( + stream: stream, gateway: gateway, session: session + ) + let running = Task { try await server.run() } + + for _ in 0..<2_000 { + if await stream.waitingReaderCount() > 0 { break } + await Task.yield() + } + #expect(await stream.waitingReaderCount() > 0) + await stream.close() + try await running.value + #expect(await gateway.sessionCount() == 0) +} + @Test func appleRVT1GatewayRunCancellationClosesConnectionAndRevokesSession() async throws { let host = TestGatewayHost() let gateway = RivetDeviceGateway(host: host) From d82e9fd5cc3ac1f61bb54a226b96530e6c733a7d Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 19:48:26 +0800 Subject: [PATCH 10/12] fix(apple): make generated RPC clients work with released SwiftPM runtime --- .../Sources/RivetHost/GeneratedBackend.swift | 34 ++++++++++++++++-- .../RivetDeviceTests/DeviceGatewayTests.swift | 3 +- .../Sources/RivetHost/GeneratedBackend.swift | 35 ++++++++++++++++--- rivet-cli/codegen/swift.rkt | 3 +- rivet/tests/codegen.rkt | 7 ++-- 5 files changed, 72 insertions(+), 10 deletions(-) diff --git a/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift b/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift index adab6b6..2c2266e 100644 --- a/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift +++ b/examples/taskboard/macos-host/Sources/RivetHost/GeneratedBackend.swift @@ -166,9 +166,39 @@ public enum RivetEvent: Sendable { } } +public protocol RivetGeneratedRPCClient: Sendable { + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue +} + +// Keep generated clients compatible with the last released SwiftPM runtime. +// Authenticated network clients can use RivetAPI(call: deviceClient.call). +public extension RivetGeneratedRPCClient { + func getState(_ name: String) async throws -> RivetValue { + try await call("$state/get", arguments: [.string(name)]) + } + + @discardableResult + func setState(_ name: String, value: RivetValue) async throws -> RivetValue { + try await call("$state/set", arguments: [.string(name), value]) + } +} + +extension RivetClient: RivetGeneratedRPCClient {} + +private struct RivetGeneratedClosureClient: RivetGeneratedRPCClient { + let invoke: @Sendable (String, [RivetValue]) async throws -> RivetValue + + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue { + try await invoke(name, arguments) + } +} + public struct RivetAPI: Sendable { - public let client: any RivetRPCClient - public init(client: any RivetRPCClient) { self.client = client } + public let client: any RivetGeneratedRPCClient + public init(client: any RivetGeneratedRPCClient) { self.client = client } + public init(call: @escaping @Sendable (String, [RivetValue]) async throws -> RivetValue) { + self.client = RivetGeneratedClosureClient(invoke: call) + } public func create_task(title: String, notes: String) async throws -> RivetTypes.BoardTask { let rivetResult = try await client.call("create-task", arguments: [encode_String(title), encode_String(notes)]) diff --git a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift index 199cce9..6668f0c 100644 --- a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift @@ -132,9 +132,10 @@ private let pairedScope = RivetDeviceScope( stateName: "sync.status", payload: payload ) - #expect(response == Data([0x42])) + #expect(response == payload) let starts = await host.starts #expect(starts.map(\.name) == ["$state/get"]) + #expect(starts.map(\.payload) == [payload]) } @Test func remoteRequestIDsMapToHostIDsAndCancellationStaysInSession() async throws { diff --git a/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift b/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift index d3d69e4..74e1842 100644 --- a/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift +++ b/platform/macos/host/Sources/RivetHost/GeneratedBackend.swift @@ -17,11 +17,38 @@ public enum RivetGeneratedConfig { public static let releaseChannel = "stable" } -public struct RivetAPI: Sendable { - public let client: any RivetRPCClient +public protocol RivetGeneratedRPCClient: Sendable { + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue +} + +// Keep generated clients compatible with the last released SwiftPM runtime. +// Authenticated network clients can use RivetAPI(call: deviceClient.call). +public extension RivetGeneratedRPCClient { + func getState(_ name: String) async throws -> RivetValue { + try await call("$state/get", arguments: [.string(name)]) + } + + @discardableResult + func setState(_ name: String, value: RivetValue) async throws -> RivetValue { + try await call("$state/set", arguments: [.string(name), value]) + } +} - public init(client: any RivetRPCClient) { - self.client = client +extension RivetClient: RivetGeneratedRPCClient {} + +private struct RivetGeneratedClosureClient: RivetGeneratedRPCClient { + let invoke: @Sendable (String, [RivetValue]) async throws -> RivetValue + + func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue { + try await invoke(name, arguments) + } +} + +public struct RivetAPI: Sendable { + public let client: any RivetGeneratedRPCClient + public init(client: any RivetGeneratedRPCClient) { self.client = client } + public init(call: @escaping @Sendable (String, [RivetValue]) async throws -> RivetValue) { + self.client = RivetGeneratedClosureClient(invoke: call) } public func greet(name: String) async throws -> String { diff --git a/rivet-cli/codegen/swift.rkt b/rivet-cli/codegen/swift.rkt index d93cea1..19db5a2 100644 --- a/rivet-cli/codegen/swift.rkt +++ b/rivet-cli/codegen/swift.rkt @@ -309,7 +309,8 @@ (apply string-append (map swift-decoder types)) "\n" (generate-swift-events events) - "public struct RivetAPI: Sendable {\n public let client: any RivetRPCClient\n public init(client: any RivetRPCClient) { self.client = client }\n\n" + "public protocol RivetGeneratedRPCClient: Sendable {\n func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue\n}\n\n// Keep generated clients compatible with the last released SwiftPM runtime.\n// Authenticated network clients can use RivetAPI(call: deviceClient.call).\npublic extension RivetGeneratedRPCClient {\n func getState(_ name: String) async throws -> RivetValue {\n try await call(\"$state/get\", arguments: [.string(name)])\n }\n\n @discardableResult\n func setState(_ name: String, value: RivetValue) async throws -> RivetValue {\n try await call(\"$state/set\", arguments: [.string(name), value])\n }\n}\n\nextension RivetClient: RivetGeneratedRPCClient {}\n\nprivate struct RivetGeneratedClosureClient: RivetGeneratedRPCClient {\n let invoke: @Sendable (String, [RivetValue]) async throws -> RivetValue\n\n func call(_ name: String, arguments: [RivetValue]) async throws -> RivetValue {\n try await invoke(name, arguments)\n }\n}\n\n" + "public struct RivetAPI: Sendable {\n public let client: any RivetGeneratedRPCClient\n public init(client: any RivetGeneratedRPCClient) { self.client = client }\n public init(call: @escaping @Sendable (String, [RivetValue]) async throws -> RivetValue) {\n self.client = RivetGeneratedClosureClient(invoke: call)\n }\n\n" (apply string-append (map swift-rpc-method rpcs)) (if (null? states) "" "\n // Shared state\n") (apply string-append (map swift-state-methods states)) diff --git a/rivet/tests/codegen.rkt b/rivet/tests/codegen.rkt index a0a8804..f2ad86d 100644 --- a/rivet/tests/codegen.rkt +++ b/rivet/tests/codegen.rkt @@ -219,8 +219,11 @@ RKT ".rivet" "generated" "kotlin" "dev" "rivet" "generated" "GeneratedBackend.kt"))) - (check-true (regexp-match? #rx"public let client: any RivetRPCClient" swift)) - (check-true (regexp-match? #rx"public init\\(client: any RivetRPCClient\\)" swift)) + (check-true (string-contains? swift "public protocol RivetGeneratedRPCClient: Sendable")) + (check-true (string-contains? swift "extension RivetClient: RivetGeneratedRPCClient {}")) + (check-true (string-contains? swift "public let client: any RivetGeneratedRPCClient")) + (check-true (string-contains? swift "public init(call: @escaping @Sendable")) + (check-false (string-contains? swift "public let client: any RivetRPCClient")) (check-true (regexp-match? #rx"public enum RivetTypes" swift)) (check-true (regexp-match? #rx" public struct User: Sendable" swift)) (check-true (regexp-match? #rx"displayName = \"demo\"" swift)) From 53715a36fb79035b24d4f86d103f31c941d49009 Mon Sep 17 00:00:00 2001 From: TuringLambdaAI Date: Thu, 22 Oct 2026 21:34:52 +0800 Subject: [PATCH 11/12] test(apple): keep reserved state event outside ordinary grants --- .../RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index c7d53a2..94abc64 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -454,7 +454,7 @@ private func helloFrame() throws -> Data { pairedDeviceID: UUID(), scopes: [RivetDeviceScope( name: "companion", - events: ["progress", "$state"], + events: ["progress"], readableStates: ["volume"] )] ) From 502a4e12bf54ba6bb6c492995290c40a16ec1eca Mon Sep 17 00:00:00 2001 From: jrtxio Date: Fri, 23 Oct 2026 07:20:15 +0800 Subject: [PATCH 12/12] fix(apple): set gateway readiness before the hello write The hello write suspends while its bytes join the channel's serialized queue. A product that observes the hello bytes and immediately calls sendEvent could have its call scheduled on the gateway actor before the post-send resumption set ready, failing with notRunning even though the wire order was still correct. Ready now precedes the write; the channel queue keeps hello ahead of any event, and a failed hello still closes the server through the existing error path. Reproduced one failure in roughly six local runs of the per-device filtering test; thirty runs pass after the fix, and the full Swift suite stays green. --- .../RivetDeviceAppleRVT1GatewayServer.swift | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift index e5eeaa2..214e4ff 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -80,8 +80,13 @@ public actor RivetDeviceAppleRVT1GatewayServer { .string("rivet"), .int64(Int64(rivetProtocolVersion)) ])) - try await channel.send(RivetFrame(type: .hello, id: 0, payload: hello)) + // The hello write suspends while its bytes join the channel's + // serialized queue, and a product observing those bytes may call + // sendEvent immediately. Ready must be set before the write so that + // call does not race this actor's resumption and fail with + // notRunning; the channel queue keeps hello ahead of any event. ready = true + try await channel.send(RivetFrame(type: .hello, id: 0, payload: hello)) while !stopping { let frame = try await channel.receive() switch frame.type {