Implement Swift request idle progress

fasterthanlime · fasterthanlime · commit b07edb7d5fcb · 2026-06-17T10:50:36.000+02:00
diff --git a/swift/vox-runtime/Sources/VoxRuntime/Channel.swift b/swift/vox-runtime/Sources/VoxRuntime/Channel.swift
@@ -636,6 +636,7 @@ public enum ChannelError: Error, Equatable {
     case reset
     case requestClosed
     case cancelled
+    case timedOut
     case connectionClosed
     case unknown
 }
diff --git a/swift/vox-runtime/Sources/VoxRuntime/Driver+Incoming.swift b/swift/vox-runtime/Sources/VoxRuntime/Driver+Incoming.swift
@@ -193,6 +193,8 @@ extension Driver {
                     channels: call.channels
                 )
             case .response(let response):
+                // r[impl rpc.request.scope.terminal]
+                // r[impl rpc.request.scope.channels]
                 // r[impl rpc.response]
                 let payload = [UInt8](response.ret)
                 guard let pending = await state.claimPendingResponse(request.id, reason: "response")
@@ -226,6 +228,8 @@ extension Driver {
                 pending.responseTx(.success(payload))
             case .cancel:
                 // r[impl rpc.cancel]
+                // r[impl rpc.request.scope.terminal]
+                // r[impl rpc.request.scope.channels]
                 let responseContext = await state.removeInFlight(request.id)
                 await terminateRequestChannels(
                     connectionId: responseContext.connectionId,
@@ -242,16 +246,32 @@ extension Driver {
                     channelId: channel.id,
                     payload: itemBytes
                 )
+                await markChannelRequestProgress(
+                    connectionId: msg.connectionId,
+                    channelId: channel.id
+                )
             case .close:
                 try await handleClose(connectionId: msg.connectionId, channelId: channel.id)
+                await markChannelRequestProgress(
+                    connectionId: msg.connectionId,
+                    channelId: channel.id
+                )
             case .reset:
                 await deliverChannelReset(connectionId: msg.connectionId, channelId: channel.id)
+                await markChannelRequestProgress(
+                    connectionId: msg.connectionId,
+                    channelId: channel.id
+                )
             case .grantCredit(let credit):
                 await deliverChannelCredit(
                     connectionId: msg.connectionId,
                     channelId: channel.id,
                     bytes: credit.additional
                 )
+                await markChannelRequestProgress(
+                    connectionId: msg.connectionId,
+                    channelId: channel.id
+                )
             }
         }
     }
@@ -423,6 +443,8 @@ extension Driver {
         channelIds: [UInt64],
         error: ChannelError
     ) async {
+        // r[impl rpc.request.scope.channels]
+        // r[impl rpc.channel.lifecycle]
         for channelId in channelIds {
             await deliverChannelReset(
                 connectionId: connectionId,
@@ -477,6 +499,11 @@ extension Driver {
 
         for (_, pending) in responses {
             pending.timeoutTask?.cancel()
+            await terminateRequestChannels(
+                connectionId: pending.request.connectionId,
+                channelIds: pending.request.channels,
+                error: .connectionClosed
+            )
             pending.responseTx(.failure(.connectionClosed))
         }
     }
@@ -489,6 +516,11 @@ extension Driver {
 
         for (_, pending) in responses {
             pending.timeoutTask?.cancel()
+            await terminateRequestChannels(
+                connectionId: pending.request.connectionId,
+                channelIds: pending.request.channels,
+                error: .connectionClosed
+            )
             pending.responseTx(.failure(.connectionClosed))
         }
     }
diff --git a/swift/vox-runtime/Sources/VoxRuntime/Driver+Outgoing.swift b/swift/vox-runtime/Sources/VoxRuntime/Driver+Outgoing.swift
@@ -102,24 +102,29 @@ extension Driver {
         let msg = queued.taskMessage
         let connectionId = queued.connectionId
         let wireMsg: Message
+        let progressChannelId: UInt64?
         switch msg {
         case .data(let channelId, let payload):
             wireMsg = messageData(channelId: channelId, item: payload, connectionId: connectionId)
+            progressChannelId = channelId
         case .close(let channelId):
             wireMsg = messageChannelClose(channelId: channelId, connectionId: connectionId)
+            progressChannelId = channelId
         case .grantCredit(let channelId, let bytes):
             wireMsg = messageCredit(
                 channelId: channelId,
                 additional: bytes,
                 connectionId: connectionId
             )
+            progressChannelId = channelId
         case .schema(let methodId, let direction, let schemas):
             wireMsg = messageSchema(
                 methodId: methodId,
                 direction: direction,
                 schemas: schemas,
                 connectionId: connectionId
             )
+            progressChannelId = nil
         case .response(let requestId, let payload, let methodId, let responseSchemaClosure):
             // Advertise the response schema at THIS sequential send point (not in the
             // concurrent dispatch task): under pipelining many responses for a method
@@ -160,8 +165,15 @@ extension Driver {
                 return
             }
             wireMsg = response
+            progressChannelId = nil
         }
         try await sendOrEnqueue(wireMsg)
+        if let progressChannelId {
+            await markChannelRequestProgress(
+                connectionId: connectionId,
+                channelId: progressChannelId
+            )
+        }
     }
 
     /// Handle a command from a lane or connection handle.
@@ -245,37 +257,12 @@ extension Driver {
                 return
             }
 
-            guard let timeout else {
-                return
-            }
-
-            let timeoutNs = Self.timeoutToNanoseconds(timeout)
-            let capturedState = state
-            let capturedConduit = conduit
-            let capturedConnectionId = connectionId
-            let timeoutTask = Task {
-                do {
-                    try await Task.sleep(nanoseconds: timeoutNs)
-                } catch {
-                    return
-                }
-                guard let pending = await capturedState.claimPendingResponse(
-                    requestId,
-                    reason: "timeout"
-                ) else {
-                    return
-                }
-                pending.timeoutTask?.cancel()
-                warnLog("request timed out request_id=\(requestId) timeout_s=\(timeout)")
-                pending.responseTx(.failure(.timeout))
-                try? await capturedConduit.send(
-                    messageCancel(requestId: requestId, connectionId: capturedConnectionId)
+            if let timeout {
+                await installRequestIdleTimeout(
+                    requestId: requestId,
+                    timeout: timeout
                 )
             }
-            let installed = await state.setPendingTimeoutTask(requestId, timeoutTask: timeoutTask)
-            if !installed {
-                timeoutTask.cancel()
-            }
         case .openLane(let settings, let metadata, let dispatcher, let responseTx):
             let isClosed = await state.isConnectionClosed()
             guard !isClosed else {
@@ -396,38 +383,12 @@ extension Driver {
 
             pendingCalls.removeFirst()
 
-            guard let timeout = call.timeout else {
-                continue
-            }
-
-            let timeoutNs = Self.timeoutToNanoseconds(timeout)
-            let capturedState = state
-            let capturedConduit = conduit
-            let capturedConnectionId = call.connectionId
-            let requestId = call.requestId
-            let timeoutTask = Task {
-                do {
-                    try await Task.sleep(nanoseconds: timeoutNs)
-                } catch {
-                    return
-                }
-                guard let pending = await capturedState.claimPendingResponse(
-                    requestId,
-                    reason: "timeout"
-                ) else {
-                    return
-                }
-                pending.timeoutTask?.cancel()
-                warnLog("request timed out request_id=\(requestId) timeout_s=\(timeout)")
-                pending.responseTx(.failure(.timeout))
-                try? await capturedConduit.send(
-                    messageCancel(requestId: requestId, connectionId: capturedConnectionId)
+            if let timeout = call.timeout {
+                await installRequestIdleTimeout(
+                    requestId: call.requestId,
+                    timeout: timeout
                 )
             }
-            let installed = await state.setPendingTimeoutTask(requestId, timeoutTask: timeoutTask)
-            if !installed {
-                timeoutTask.cancel()
-            }
         }
     }
 
@@ -450,4 +411,63 @@ extension Driver {
             pendingTaskMessages.removeFirst()
         }
     }
+
+    // r[impl rpc.timeout.idle-progress]
+    func installRequestIdleTimeout(
+        requestId: UInt64,
+        timeout: TimeInterval
+    ) async {
+        let timeoutNs = Self.timeoutToNanoseconds(timeout)
+        let timeoutTask = Task { [weak self] in
+            do {
+                try await Task.sleep(nanoseconds: timeoutNs)
+            } catch {
+                return
+            }
+            await self?.expireRequestForIdleTimeout(requestId: requestId, timeout: timeout)
+        }
+        let replacement = await state.replacePendingTimeoutTask(
+            requestId,
+            timeoutTask: timeoutTask
+        )
+        guard replacement.installed else {
+            timeoutTask.cancel()
+            return
+        }
+        replacement.previous?.cancel()
+    }
+
+    // r[impl rpc.timeout.idle-progress]
+    func markChannelRequestProgress(connectionId: UInt64, channelId: UInt64) async {
+        let contexts = await state.pendingTimeoutContexts(
+            connectionId: connectionId,
+            channelId: channelId
+        )
+        for context in contexts {
+            await installRequestIdleTimeout(
+                requestId: context.requestId,
+                timeout: context.timeout
+            )
+        }
+    }
+
+    // r[impl rpc.timeout.idle-progress]
+    // r[impl rpc.request.scope.terminal]
+    // r[impl rpc.request.scope.channels]
+    func expireRequestForIdleTimeout(requestId: UInt64, timeout: TimeInterval) async {
+        guard let pending = await state.claimPendingResponse(requestId, reason: "timeout") else {
+            return
+        }
+        pending.timeoutTask?.cancel()
+        warnLog("request timed out request_id=\(requestId) timeout_s=\(timeout)")
+        await terminateRequestChannels(
+            connectionId: pending.request.connectionId,
+            channelIds: pending.request.channels,
+            error: .timedOut
+        )
+        pending.responseTx(.failure(.timeout))
+        try? await conduit.send(
+            messageCancel(requestId: requestId, connectionId: pending.request.connectionId)
+        )
+    }
 }
diff --git a/swift/vox-runtime/Sources/VoxRuntime/DriverState.swift b/swift/vox-runtime/Sources/VoxRuntime/DriverState.swift
@@ -1,5 +1,10 @@
 import Foundation
 
+struct PendingTimeoutContext: Sendable {
+    let requestId: UInt64
+    let timeout: TimeInterval
+}
+
 /// Actor that holds mutable driver state to avoid NSLock in async contexts.
 actor DriverState {
     private let retainFinalizedRequests = true
@@ -106,15 +111,52 @@ actor DriverState {
         }
     }
 
-    func setPendingTimeoutTask(_ requestId: UInt64, timeoutTask: Task<Void, Never>) -> Bool {
+    func replacePendingTimeoutTask(
+        _ requestId: UInt64,
+        timeoutTask: Task<Void, Never>
+    ) -> (installed: Bool, previous: Task<Void, Never>?) {
         guard var pending = pendingResponses[requestId] else {
-            return false
+            return (false, nil)
         }
+        let previous = pending.timeoutTask
         pending.timeoutTask = timeoutTask
         pendingResponses[requestId] = pending
-        return true
+        return (true, previous)
+    }
+
+    // r[impl rpc.timeout.idle-progress]
+    func pendingTimeoutContext(_ requestId: UInt64) -> PendingTimeoutContext? {
+        guard let pending = pendingResponses[requestId],
+            let timeout = pending.request.timeout
+        else {
+            return nil
+        }
+        return PendingTimeoutContext(
+            requestId: requestId,
+            timeout: timeout
+        )
+    }
+
+    // r[impl rpc.timeout.idle-progress]
+    func pendingTimeoutContexts(
+        connectionId: UInt64,
+        channelId: UInt64
+    ) -> [PendingTimeoutContext] {
+        pendingResponses.compactMap { requestId, pending in
+            guard pending.request.connectionId == connectionId,
+                pending.request.channels.contains(channelId),
+                let timeout = pending.request.timeout
+            else {
+                return nil
+            }
+            return PendingTimeoutContext(
+                requestId: requestId,
+                timeout: timeout
+            )
+        }
     }
 
+    // r[impl rpc.request.scope]
     func addInFlight(
         _ requestId: UInt64,
         connectionId: UInt64,
@@ -145,6 +187,7 @@ actor DriverState {
         return .inserted
     }
 
+    // r[impl rpc.request.scope.terminal]
     func removeInFlight(_ requestId: UInt64) -> (
         removed: Bool,
         connectionId: UInt64,
diff --git a/swift/vox-runtime/Tests/VoxRuntimeTests/ConnectionFailureTests.swift b/swift/vox-runtime/Tests/VoxRuntimeTests/ConnectionFailureTests.swift

Original file line number	Diff line number	Diff line change
`@@ -636,6 +636,7 @@ public enum ChannelError: Error, Equatable {`
`636`	`636`	`case reset`
`637`	`637`	`case requestClosed`
`638`	`638`	`case cancelled`
	`639`	`+ case timedOut`
`639`	`640`	`case connectionClosed`
`640`	`641`	`case unknown`
`641`	`642`	`}`