fix(device): sweep wedged dispatches, bound the rvt1 hello, and pin the state deny surface (#312)

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)
         }

platform/macos/Sources/RivetDeviceAppleTransport/RivetDeviceAppleRVT1Client.swift

@@ -1,12 +1,15 @@
 import Foundation
 import RivetDevice
+import RivetDeviceSession
 import RivetRuntime
 
 public enum RivetDeviceAppleRVT1ClientError: Error, Equatable, Sendable {
     case alreadyStarted
     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<Void, Never>] = []
+    // 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.

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<Void, Never>
+}
+
 /// 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<UInt64>()
-    private var cancelledIDs = Set<UInt64>()
-    private var requestTasks: [UInt64: Task<Void, Never>] = [:]
+    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))

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 {