Make XPCClient.send timeout actually cancel pending replies

The previous implementation raced `Task.sleep` against the XPC reply
inside a `withThrowingTaskGroup`. When the timeout won, structured
concurrency required the group to await the XPC child task before the
group scope could return — but that child was suspended in a
`withCheckedThrowingContinuation` that only resumes when the C
`xpc_connection_send_message_with_reply` callback fires. Cancelling a
Swift Task does not cancel the underlying C call. If the remote service
was wedged (no reply, no connection invalidation), the child never
resumed and the group never returned, regardless of the supplied
`responseTimeout`. The `responseTimeout` parameter was therefore
silently ineffective in exactly the failure mode it was meant to
mitigate.

This commit replaces the TaskGroup with a single-resume gate
(`ResumptionState`) over a `CheckedContinuation` wrapped in a
`withTaskCancellationHandler`. The continuation is resumed by whichever
of the following completes first:

  1. The XPC reply callback fires.
  2. `responseTimeout` elapses.
  3. The current Task is cancelled.

Late completions from the other paths are dropped silently, so the
underlying XPC connection remains valid for subsequent sends. This is
required for callers that hold a long-lived `XPCClient`
(`ContainerClient`, `NetworkClient`); a simpler design that called
`xpc_connection_cancel` on timeout would brick those clients after a
single timed-out send.

Tradeoffs documented in docs/internal/help-freeze-analysis.md:
  - On timeout/cancel, the eventual late XPC reply is retained by XPC
    until the connection is released. For short-lived clients this is
    GC'd within milliseconds; for long-lived reusable clients the worst
    case is one orphaned `xpc_object_t` per timed-out send.
  - The unstructured `Task` that runs the timeout sleep is not
    cancelled when the parent task is cancelled; it wakes up later and
    becomes a no-op via `tryResume`.

Reviewers: a unit test that injects a connection with a non-firing reply
would meaningfully cover both the timeout path and the reusable-client
guarantee. Happy to add it in this PR or as a follow-up — preference?

Author: chrisgeo

Sources/ContainerXPC/XPCClient.swift

@@ -70,44 +70,59 @@ extension XPCClient {
     }
 
     /// Send the provided message to the service.
+    ///
+    /// The response is delivered by whichever of the following completes first:
+    ///   1. The XPC reply callback fires.
+    ///   2. `responseTimeout` elapses.
+    ///   3. The current `Task` is cancelled.
+    ///
+    /// Late completions from the other paths are dropped silently so the connection
+    /// remains valid for subsequent sends — important for callers that hold a long-lived
+    /// `XPCClient` (`ContainerClient`, `NetworkClient`). A previous implementation used
+    /// a `withThrowingTaskGroup`, but structured-concurrency cleanup awaited the XPC
+    /// child task, which could not actually be cancelled because
+    /// `xpc_connection_send_message_with_reply` only resumes when its callback fires.
+    /// That made `responseTimeout` ineffective whenever the remote service was wedged.
+    /// See docs/internal/help-freeze-analysis.md.
     @discardableResult
     public func send(_ message: XPCMessage, responseTimeout: Duration? = nil) async throws -> XPCMessage {
-        try await withThrowingTaskGroup(of: XPCMessage.self, returning: XPCMessage.self) { group in
-            if let responseTimeout {
-                group.addTask {
-                    try await Task.sleep(for: responseTimeout)
-                    let route = message.string(key: XPCMessage.routeKey) ?? "nil"
-                    throw ContainerizationError(
-                        .internalError,
-                        message: "XPC timeout for request to \(self.service)/\(route)"
-                    )
+        let state = ResumptionState<XPCMessage>()
+        return try await withTaskCancellationHandler {
+            try await withCheckedThrowingContinuation { (cont: CheckedContinuation<XPCMessage, Error>) in
+                state.set(cont)
+
+                xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+                    do {
+                        let parsed = try self.parseReply(reply)
+                        state.tryResume(returning: parsed)
+                    } catch {
+                        state.tryResume(throwing: error)
+                    }
                 }
-            }
 
-            group.addTask {
-                try await withCheckedThrowingContinuation { cont in
-                    xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
-                        do {
-                            let message = try self.parseReply(reply)
-                            cont.resume(returning: message)
-                        } catch {
-                            cont.resume(throwing: error)
-                        }
+                if let responseTimeout {
+                    let service = self.service
+                    let route = message.string(key: XPCMessage.routeKey) ?? "nil"
+                    Task { [state] in
+                        try? await Task.sleep(for: responseTimeout)
+                        state.tryResume(
+                            throwing: ContainerizationError(
+                                .timeout,
+                                message: "XPC timeout for request to \(service)/\(route)"
+                            )
+                        )
                     }
                 }
-            }
-
-            let response = try await group.next()
-            // once one task has finished, cancel the rest.
-            group.cancelAll()
-            // we don't really care about the second error here
-            // as it's most likely a `CancellationError`.
-            try? await group.waitForAll()
 
-            guard let response else {
-                throw ContainerizationError(.invalidState, message: "failed to receive XPC response")
+                // Close the race window: if cancellation arrived before `set(cont)`
+                // ran, the cancellation handler resumed against an empty state. Resume
+                // here so the continuation cannot be lost.
+                if Task.isCancelled {
+                    state.tryResume(throwing: CancellationError())
+                }
             }
-            return response
+        } onCancel: {
+            state.tryResume(throwing: CancellationError())
         }
     }
 
@@ -133,4 +148,36 @@ extension XPCClient {
     }
 }
 
+/// Single-resume gate around a `CheckedContinuation`.
+///
+/// `XPCClient.send` races multiple completion sources (XPC reply callback, timeout
+/// sleep, parent-task cancellation) against a single continuation. Whichever wins
+/// resumes via `tryResume`; subsequent resumes are dropped silently.
+private final class ResumptionState<T: Sendable>: @unchecked Sendable {
+    private let lock = NSLock()
+    private var continuation: CheckedContinuation<T, Error>?
+
+    func set(_ continuation: CheckedContinuation<T, Error>) {
+        lock.lock()
+        defer { lock.unlock() }
+        self.continuation = continuation
+    }
+
+    func tryResume(returning value: T) {
+        lock.lock()
+        let c = continuation
+        continuation = nil
+        lock.unlock()
+        c?.resume(returning: value)
+    }
+
+    func tryResume(throwing error: Error) {
+        lock.lock()
+        let c = continuation
+        continuation = nil
+        lock.unlock()
+        c?.resume(throwing: error)
+    }
+}
+
 #endif