From d8d09fd53232dca3afd9412ddd37410c76e43f67 Mon Sep 17 00:00:00 2001 From: jrtxio Date: Sun, 25 Oct 2026 11:08:14 +0800 Subject: [PATCH] fix(device): sweep wedged dispatches, bound the rvt1 hello, and pin the state deny surface (#312) Follow-ups from the #305 gateway review: - Hello read deadline: the rvt1 client's post-authentication hello read carried no budget, so a silent peer held the session forever. It now mirrors the rvd1 post-TLS deadline policy (watchdog closes the channel; a racing deadline cannot close a session that already holds the hello). - Typed handshake failures: pre-hello transport and protocol errors surface as handshakeTimedOut / handshakeFailed instead of raw channel errors. - Wedged-backend capacity: cancelling a request now frees its wire slot immediately and retires the dispatch by token, so a backend task that ignores cancellation cannot keep the slot or answer a reused id; the gateway session likewise reclaims the pending entry at cancel (its vanished-entry rule already suppresses the late response). The backend host drops its work entry at cancel too. - Deny surface: the gateway tests now pin $state/set denial through the ordinary RPC route, both state routes against ungranted names, the write-capability allow path, and an end-to-end $state/set denial. Red-green: the capacity-reclaim test fails against the pre-fix server (rejection error frame) and passes with the fix; the full swift test run is green. --- .../Sources/RivetDevice/DeviceGateway.swift | 13 +- .../RivetDeviceAppleRVT1Client.swift | 59 ++++++++- .../RivetDeviceAppleRVT1GatewayServer.swift | 56 +++++---- .../RivetDeviceRVT1BackendHost.swift | 5 +- .../AppleRVT1GatewayTests.swift | 113 +++++++++++++++++- .../RivetDeviceTests/DeviceGatewayTests.swift | 36 +++++- 6 files changed, 242 insertions(+), 40 deletions(-) diff --git a/platform/macos/Sources/RivetDevice/DeviceGateway.swift b/platform/macos/Sources/RivetDevice/DeviceGateway.swift index 0c89b41c..48f439e9 100644 --- a/platform/macos/Sources/RivetDevice/DeviceGateway.swift +++ b/platform/macos/Sources/RivetDevice/DeviceGateway.swift @@ -292,14 +292,15 @@ public actor RivetDeviceGatewaySession { /// Cancellation is deliberately silent for unknown IDs so one session /// cannot probe request ownership in another session. public func cancel(remoteRequestID: UInt64) async { - guard var entry = pending[remoteRequestID] else { + // The pending slot is reclaimed immediately: a host that ignores the + // cancellation would otherwise pin this entry — and the session's + // pending capacity — until the connection ends. A completion arriving + // after removal still reports cancellation (cancellationWasRequested + // treats a vanished entry as cancelled), and a second cancel for the + // same id finds no entry and stays silent. + guard let entry = pending.removeValue(forKey: remoteRequestID) else { return } - guard !entry.cancellationRequested else { - return - } - entry.cancellationRequested = true - pending[remoteRequestID] = entry if let hostRequestID = entry.hostRequestID { await host.cancelRPC(id: hostRequestID) } diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift index 2e5450ea..4d76525e 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift @@ -1,5 +1,6 @@ import Foundation import RivetDevice +import RivetDeviceSession import RivetRuntime public enum RivetDeviceAppleRVT1ClientError: Error, Equatable, Sendable { @@ -7,6 +8,8 @@ public enum RivetDeviceAppleRVT1ClientError: Error, Equatable, Sendable { case invalidHello case invalidResponse case invalidState + case handshakeTimedOut + case handshakeFailed(String) case tooManyPendingRequests(Int) case messageTooLarge(actual: Int, maximum: Int) case remote(String) @@ -25,6 +28,7 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { public typealias EventHandler = @Sendable (String, RivetValue) -> Void private let channel: RivetDeviceAppleRVT1Channel private let maximumPendingRequests: Int + private let helloTimeout: TimeInterval private var nextRequestID: UInt64 = 1 private var pending: [UInt64: RivetDeviceApplePendingRequest] = [:] private var started = false @@ -34,25 +38,48 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { private var terminated = false private var terminalError: (any Error)? private var terminationWaiters: [CheckedContinuation] = [] + // Guards the one-shot hello deadline: the watchdog fires only when this + // token still names the in-flight handshake attempt. + private var helloAttempt = 0 + private var helloTimedOut = false public init(connection: RivetDeviceAppleConnection, - maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests) { + maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests, + helloTimeout: TimeInterval = rivetDeviceSessionPostTLSTimeout) { precondition(maximumPendingRequests > 0) + precondition(helloTimeout.isFinite && helloTimeout > 0) channel = RivetDeviceAppleRVT1Channel(connection: connection) self.maximumPendingRequests = maximumPendingRequests + self.helloTimeout = helloTimeout } package init(stream: any RivetDeviceAppleApplicationByteStream, - maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests) { + maximumPendingRequests: Int = rivetDeviceDefaultMaximumPendingRequests, + helloTimeout: TimeInterval = rivetDeviceSessionPostTLSTimeout) { precondition(maximumPendingRequests > 0) + precondition(helloTimeout.isFinite && helloTimeout > 0) channel = RivetDeviceAppleRVT1Channel(stream: stream) self.maximumPendingRequests = maximumPendingRequests + self.helloTimeout = helloTimeout } public func start(onEvent: EventHandler? = nil) async throws { guard !started, !stopped else { throw RivetDeviceAppleRVT1ClientError.alreadyStarted } + // A peer that RVD1 authenticated but then goes silent must not hold + // the session open forever: the hello read carries the same post-TLS + // deadline policy the RVD1 session layer applies (issue #312). + helloTimedOut = false + helloAttempt += 1 + let attempt = helloAttempt + let timeout = helloTimeout + let watchdog = Task { [weak self] in + try? await Task.sleep(for: .seconds(timeout)) + guard !Task.isCancelled else { return } + await self?.expireHello(unlessAttemptIs: attempt) + } + defer { watchdog.cancel() } let hello: RivetFrame do { hello = try await withTaskCancellationHandler { @@ -61,11 +88,19 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { Task { await self.stop() } } } catch { + helloAttempt += 1 stopped = true await channel.close() if Task.isCancelled { throw CancellationError() } - throw error + if helloTimedOut { + throw RivetDeviceAppleRVT1ClientError.handshakeTimedOut + } + throw Self.typedHandshakeError(error) } + // Invalidate the watchdog before any further await so a deadline + // that raced the hello cannot close a session that already holds it. + helloAttempt += 1 + watchdog.cancel() guard hello.type == .hello, hello.id == 0 else { stopped = true await channel.close() @@ -87,6 +122,24 @@ public actor RivetDeviceAppleRVT1Client: RivetRPCClient { readerTask = Task { [weak self] in await self?.readLoop() } } + /// Closes the channel so the suspended hello read fails promptly, and + /// marks the handshake as deadline-expired for the catch above. + private func expireHello(unlessAttemptIs attempt: Int) async { + guard helloAttempt == attempt, !stopped else { return } + helloAttempt += 1 + helloTimedOut = true + await channel.close() + } + + private static func typedHandshakeError( + _ error: any Error + ) -> RivetDeviceAppleRVT1ClientError { + if let clientError = error as? RivetDeviceAppleRVT1ClientError { + return clientError + } + return .handshakeFailed(String(describing: error)) + } + /// 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. diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift index 214e4ff9..c4d9e3cc 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1GatewayServer.swift @@ -11,6 +11,15 @@ public enum RivetDeviceAppleRVT1GatewayServerError: Error, Equatable, Sendable { case duplicateRequestID(UInt64) } +/// Identity for one request dispatch. A backend that ignores cancellation +/// never completes its task, so completions are matched by per-dispatch +/// token: a late dispatch for a retired request id cannot answer a newer +/// request that reused the id (issue #312). +private struct RequestDispatch: Sendable { + let token: UUID + let task: Task +} + /// Terminates the post-RVD1 RVT1 stream and applies per-pairing gateway /// permissions before dispatching any RPC to the embedded backend. public actor RivetDeviceAppleRVT1GatewayServer { @@ -28,8 +37,7 @@ public actor RivetDeviceAppleRVT1GatewayServer { private var stopping = false private var closed = false private var activeIDs = Set() - private var cancelledIDs = Set() - private var requestTasks: [UInt64: Task] = [:] + private var requestTasks: [UInt64: RequestDispatch] = [:] public init( connection: RivetDeviceAppleConnection, @@ -160,12 +168,11 @@ public actor RivetDeviceAppleRVT1GatewayServer { 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() + for dispatch in tasks { + dispatch.task.cancel() } await channel.close() await gateway.closeSession(id: session.id) @@ -209,8 +216,12 @@ public actor RivetDeviceAppleRVT1GatewayServer { } let payload = try encodeRivetValue(.list(arguments)) let session = self.session - let task = Task { [weak self] in + let token = UUID() + let dispatch = RequestDispatch(token: token, task: Task { [weak self] in do { + // A dispatch cancelled before it starts is stopped by + // the session's own isCancelled guard, so it never + // registers a pending call. let response: Data switch route { case .rpc: @@ -227,12 +238,16 @@ public actor RivetDeviceAppleRVT1GatewayServer { payload: payload ) } - await self?.completeRequest(id: frame.id, response: response) + await self?.completeRequest( + id: frame.id, token: token, response: response + ) } catch { - await self?.completeRequest(id: frame.id, error: error) + await self?.completeRequest( + id: frame.id, token: token, error: error + ) } - } - requestTasks[frame.id] = task + }) + requestTasks[frame.id] = dispatch } catch { activeIDs.remove(frame.id) try await sendError(id: frame.id, message: safeMessage(for: error)) @@ -241,27 +256,24 @@ public actor RivetDeviceAppleRVT1GatewayServer { 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() + // The peer retired this id when it sent the cancel, so the wire slot + // is freed immediately; a dispatch that ignores cancellation keeps + // running but its late completion no longer matches (issue #312). + activeIDs.remove(id) + requestTasks.removeValue(forKey: id)?.task.cancel() await session.cancel(remoteRequestID: id) } private func completeRequest( id: UInt64, + token: UUID, 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) + guard requestTasks[id]?.token == token else { return } requestTasks.removeValue(forKey: id) + activeIDs.remove(id) + guard !closed else { return } do { if let response { try await channel.send(RivetFrame(type: .response, id: id, payload: response)) diff --git a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift index ef94cd04..8de96d27 100644 --- a/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift +++ b/platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceRVT1BackendHost.swift @@ -46,7 +46,10 @@ public actor RivetDeviceRVT1BackendHost: RivetDeviceGatewayHost { } public func cancelRPC(id: UInt64) async { - work[id]?.cancel() + // Drop the entry now: a backend call that ignores cancellation never + // reaches finish(), and the abandoned response task no-ops when it + // removes an id that is already gone. + work.removeValue(forKey: id)?.cancel() } private func allocateID() throws -> UInt64 { diff --git a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift index 94abc64e..ad0d4409 100644 --- a/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceAppleTransportTests/AppleRVT1GatewayTests.swift @@ -50,6 +50,10 @@ private actor MemoryByteStream: RivetDeviceAppleApplicationByteStream { try sent.map { try decodeRivetFrame($0) } } + func errorFrameCount() throws -> Int { + try sentFrames().filter { $0.type == .error }.count + } + private func take(_ maximumLength: Int) -> Data { let count = min(maximumLength, incoming.count) let result = Data(incoming.prefix(count)) @@ -69,6 +73,27 @@ private actor MemoryByteStream: RivetDeviceAppleApplicationByteStream { } } +/// Models a backend dispatch that never completes and ignores cancellation: +/// the exact shape that used to permanently shrink the gateway's request +/// capacity (issue #312). +private actor WedgedGatewayHost: RivetDeviceGatewayHost { + private var starts = 0 + + func startRPC(name: String, payload: Data) async throws -> RivetDeviceGatewayHostCall { + starts += 1 + let task = Task { + try await withCheckedThrowingContinuation { + (_: CheckedContinuation) in + // Never resumed: cancellation is deliberately ignored. + } + } + return RivetDeviceGatewayHostCall(id: UInt64(starts), response: task) + } + + func cancelRPC(id: UInt64) async {} + func startCount() -> Int { starts } +} + private actor SlowGatewayHost: RivetDeviceGatewayHost { private var starts = 0 private var cancellations = 0 @@ -192,6 +217,34 @@ private func helloFrame() throws -> Data { #expect(await stream.isClosed()) } +@Test func appleRVT1ClientTimesOutASilentPeerWithTheHelloDeadline() async throws { + let stream = MemoryByteStream() + let client = RivetDeviceAppleRVT1Client(stream: stream, helloTimeout: 0.2) + await #expect(throws: RivetDeviceAppleRVT1ClientError.handshakeTimedOut) { + try await client.start() + } + #expect(await stream.isClosed()) +} + +@Test func appleRVT1ClientTypesPreHelloTransportFailures() async throws { + // Valid magic and version, unknown message type: frame decoding fails + // before any hello, and the raw protocol error must surface through the + // typed client error instead (issue #312). + let garbage = Data([0x52, 0x56, 0x54, 0x31, rivetProtocolVersion, 0xFF]) + Data(count: 12) + let stream = MemoryByteStream(incoming: garbage) + let client = RivetDeviceAppleRVT1Client(stream: stream) + do { + try await client.start() + Issue.record("the handshake must fail on an undecodable frame") + } catch let error as RivetDeviceAppleRVT1ClientError { + guard case .handshakeFailed = error else { + Issue.record("unexpected error: \(error)") + return + } + } + #expect(await stream.isClosed()) +} + @Test func appleRVT1ClientRunOwnsConnectionLifetimeUntilStopped() async throws { let stream = MemoryByteStream(incoming: try helloFrame()) let client = RivetDeviceAppleRVT1Client(stream: stream) @@ -262,6 +315,12 @@ 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")]))), + // End-to-end write denial: the scope reads "volume" but cannot + // write any state (issue #312). + RivetFrame(type: .request, id: 10, + payload: try encodeRivetValue(.list([ + .string("$state/set"), .string("volume"), .int64(11) + ]))), ] for frame in frames { incoming.append(try encodedFrame(frame)) } let stream = MemoryByteStream(incoming: incoming) @@ -270,10 +329,10 @@ private func helloFrame() throws -> Data { ) let running = Task { try await server.run() } for _ in 0..<2_000 { - if await stream.sentCount() >= 4 { break } + if await stream.sentCount() >= 5 { break } await Task.yield() } - #expect(await stream.sentCount() >= 4) + #expect(await stream.sentCount() >= 5) await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) try await running.value @@ -285,8 +344,11 @@ private func helloFrame() throws -> Data { #expect(byID[7]?.type == .response) #expect(byID[8]?.type == .response) #expect(byID[9]?.type == .error) + #expect(byID[10]?.type == .error) let denied = try #require(byID[9]) #expect(try decodeRivetValue(denied.payload) == .string("operation not permitted")) + let writeDenied = try #require(byID[10]) + #expect(try decodeRivetValue(writeDenied.payload) == .string("operation not permitted")) #expect(Set(await host.names()) == Set(["player.play", "$state/get"])) #expect(await gateway.sessionCount() == 0) } @@ -364,6 +426,53 @@ private func helloFrame() throws -> Data { #expect(await gateway.sessionCount() == 0) } +@Test func appleRVT1GatewayReclaimsCancelledSlotsWhenBackendIgnoresCancellation() async throws { + let host = WedgedGatewayHost() + let gateway = RivetDeviceGateway(host: host) + let session = try await gateway.openSession( + pairedDeviceID: UUID(), + scopes: [RivetDeviceScope(name: "companion", rpcs: ["player.play"])] + ) + // Fill the pending capacity with dispatches that never complete, then + // cancel one: its wire slot must return to the pool even though the + // wedged backend task keeps running (issue #312). + var incoming = Data() + for id in 1...rivetDeviceDefaultMaximumPendingRequests { + incoming.append(try encodedFrame(RivetFrame( + type: .request, + id: UInt64(id), + payload: try encodeRivetValue(.list([.string("player.play")])) + ))) + } + incoming.append(try encodedFrame(RivetFrame(type: .cancel, id: 1))) + incoming.append(try encodedFrame(RivetFrame( + type: .request, + id: UInt64(rivetDeviceDefaultMaximumPendingRequests + 1), + 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 { + let starts = await host.startCount() + let errors = (try? await stream.errorFrameCount()) ?? 1 + if starts >= rivetDeviceDefaultMaximumPendingRequests + 1 || errors > 0 { + break + } + await Task.yield() + } + #expect(await host.startCount() >= rivetDeviceDefaultMaximumPendingRequests) + + await stream.enqueue(try encodedFrame(RivetFrame(type: .shutdown, id: 0))) + try await running.value + let sent = try await stream.sentFrames() + #expect(!sent.contains { $0.type == .error }) + #expect(await gateway.sessionCount() == 0) +} + @Test func appleRVT1GatewayShutdownCancelsInFlightRPCAndRevokesSession() async throws { let host = SlowGatewayHost() let gateway = RivetDeviceGateway(host: host) diff --git a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift index 6668f0c6..60164008 100644 --- a/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift +++ b/platform/macos/Tests/RivetDeviceTests/DeviceGatewayTests.swift @@ -110,6 +110,8 @@ private let pairedScope = RivetDeviceScope( ) let payload = Data("encoded state args".utf8) + // State operations are only reachable through invokeStateRPC; the + // ordinary RPC route denies both directions. await #expect(throws: RivetDeviceGatewayError.forbidden) { try await session.invokeRPC( remoteRequestID: 20, @@ -118,24 +120,46 @@ private let pairedScope = RivetDeviceScope( ) } await #expect(throws: RivetDeviceGatewayError.forbidden) { - try await session.invokeStateRPC( + try await session.invokeRPC( remoteRequestID: 21, + name: "$state/set", + payload: payload + ) + } + await #expect(throws: RivetDeviceGatewayError.forbidden) { + try await session.invokeStateRPC( + remoteRequestID: 22, name: "$state/get", stateName: "private.state", payload: payload ) } + await #expect(throws: RivetDeviceGatewayError.forbidden) { + try await session.invokeStateRPC( + remoteRequestID: 23, + name: "$state/set", + stateName: "private.state", + payload: payload + ) + } - let response = try await session.invokeStateRPC( - remoteRequestID: 22, + let read = try await session.invokeStateRPC( + remoteRequestID: 24, name: "$state/get", stateName: "sync.status", payload: payload ) - #expect(response == payload) + #expect(read == payload) + let write = try await session.invokeStateRPC( + remoteRequestID: 25, + name: "$state/set", + stateName: "sync.preference", + payload: payload + ) + #expect(write == payload) let starts = await host.starts - #expect(starts.map(\.name) == ["$state/get"]) - #expect(starts.map(\.payload) == [payload]) + #expect(starts.map(\.name) == ["$state/get", "$state/set"]) + #expect(starts.map(\.payload) == [payload, payload]) } @Test func remoteRequestIDsMapToHostIDsAndCancellationStaysInSession() async throws {