fix(xpc): honor request timeouts

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif

Package.swift

@@ -478,6 +478,12 @@ let package = Package(
  "CAuditToken",
  ]
  ),
+ .testTarget(
+ name: "ContainerXPCTests",
+ dependencies: [
+ "ContainerXPC"
+ ]
+ ),
  .target(
  name: "ContainerOS",
  dependencies: [

Sources/ContainerXPC/XPCClient.swift

@@ -111,15 +111,18 @@ extension XPCClient {
  }
 
  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)
+ let box = XPCReplyBox()
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { cont in
+ box.store(cont)
+ xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in
+ box.resume { try self.parseReply(reply) }
  }
  }
+ } onCancel: {
+ // XPC reply callbacks do not observe task cancellation, so
+ // resume the awaiting continuation when a timeout wins.
+ box.resume { throw CancellationError() }
  }
  }
 

Sources/ContainerXPC/XPCReplyBox.swift

@@ -0,0 +1,54 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import Foundation
+
+/// Resumes an XPC reply continuation once when either a reply or cancellation arrives.
+final class XPCReplyBox: @unchecked Sendable {
+ private let lock = NSLock()
+ private var continuation: CheckedContinuation<XPCMessage, Error>?
+ private var result: Result<XPCMessage, Error>?
+
+ func store(_ continuation: CheckedContinuation<XPCMessage, Error>) {
+ lock.lock()
+ guard let result else {
+ self.continuation = continuation
+ lock.unlock()
+ return
+ }
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+
+ func resume(_ body: () throws -> XPCMessage) {
+ let result = Result { try body() }
+ lock.lock()
+ guard self.result == nil else {
+ lock.unlock()
+ return
+ }
+ self.result = result
+ guard let continuation else {
+ lock.unlock()
+ return
+ }
+ self.continuation = nil
+ lock.unlock()
+ continuation.resume(with: result)
+ }
+}
+#endif

Tests/ContainerXPCTests/XPCClientTests.swift

@@ -0,0 +1,230 @@
+//===----------------------------------------------------------------------===//
+// Copyright © 2026 Apple Inc. and the container project authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//===----------------------------------------------------------------------===//
+
+#if os(macOS)
+import ContainerizationError
+@preconcurrency import Foundation
+import Testing
+
+@testable import ContainerXPC
+
+@Suite(.timeLimit(.minutes(1)), .serialized)
+struct XPCClientTests {
+ @Test
+ func replyBoxDeliversCancellationThatPrecedesContinuationStorage() async throws {
+ let box = XPCReplyBox()
+ box.resume { throw CancellationError() }
+
+ do {
+ _ = try await withCheckedThrowingContinuation { continuation in
+ box.store(continuation)
+ }
+ Issue.record("expected the pending cancellation to be delivered")
+ } catch is CancellationError {
+ // Expected.
+ }
+ }
+
+ @Test
+ func responseTimeoutReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let clock = ContinuousClock()
+ let start = clock.now
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang"))
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func callerCancellationReturnsWithinBound() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+ let request = Task {
+ try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .seconds(30)
+ )
+ }
+
+ try await Task.sleep(for: .milliseconds(50))
+ let clock = ContinuousClock()
+ let start = clock.now
+ request.cancel()
+
+ do {
+ _ = try await request.value
+ Issue.record("expected send to be cancelled")
+ } catch is CancellationError {
+ #expect(start.duration(to: clock.now) < .seconds(2))
+ }
+ }
+
+ @Test
+ func clientCanBeReusedAfterResponseTimeout() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+
+ @Test
+ func lateReplyAfterTimeoutIsIgnored() async throws {
+ let server = AnonymousXPCServer()
+ defer { server.close() }
+
+ let client = server.makeClient()
+
+ do {
+ _ = try await client.send(
+ XPCMessage(route: "hang"),
+ responseTimeout: .milliseconds(100)
+ )
+ Issue.record("expected send to time out")
+ } catch let error as ContainerizationError {
+ #expect(error.message.contains("XPC timeout"))
+ }
+
+ #expect(server.replyToPendingRequests())
+ try await Task.sleep(for: .milliseconds(50))
+
+ let response = try await client.send(
+ XPCMessage(route: "echo"),
+ responseTimeout: .seconds(1)
+ )
+ #expect(response.string(key: "result") == "ok")
+ }
+}
+
+private final class AnonymousXPCServer: @unchecked Sendable {
+ private let listener: xpc_connection_t
+ private let lock = NSLock()
+ private var connections = [xpc_connection_t]()
+ private var pendingRequests = [xpc_object_t]()
+
+ init() {
+ listener = xpc_connection_create(nil, nil)
+ xpc_connection_set_event_handler(listener) { [weak self] object in
+ switch xpc_get_type(object) {
+ case XPC_TYPE_CONNECTION:
+ self?.accept(connection: object)
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(listener)
+ }
+
+ func makeClient() -> XPCClient {
+ let endpoint = xpc_endpoint_create(listener)
+ let connection = xpc_connection_create_from_endpoint(endpoint)
+ return XPCClient(connection: connection, label: "test.container.xpc")
+ }
+
+ func close() {
+ xpc_connection_cancel(listener)
+ lock.withLock {
+ for connection in connections {
+ xpc_connection_cancel(connection)
+ }
+ connections.removeAll()
+ pendingRequests.removeAll()
+ }
+ }
+
+ func replyToPendingRequests() -> Bool {
+ let requests = lock.withLock {
+ let requests = pendingRequests
+ pendingRequests.removeAll()
+ return requests
+ }
+ guard let connection = lock.withLock({ connections.last }) else {
+ return false
+ }
+ for request in requests {
+ guard let reply = xpc_dictionary_create_reply(request) else {
+ continue
+ }
+ xpc_dictionary_set_string(reply, "result", "late")
+ xpc_connection_send_message(connection, reply)
+ }
+ return !requests.isEmpty
+ }
+
+ private func accept(connection: xpc_connection_t) {
+ lock.withLock {
+ connections.append(connection)
+ }
+ nonisolated(unsafe) let connection = connection
+ xpc_connection_set_event_handler(connection) { [weak self] object in
+ guard let self else {
+ return
+ }
+ switch xpc_get_type(object) {
+ case XPC_TYPE_DICTIONARY:
+ let message = XPCMessage(object: object)
+ switch message.string(key: XPCMessage.routeKey) {
+ case "echo":
+ guard let reply = xpc_dictionary_create_reply(object) else {
+ return
+ }
+ xpc_dictionary_set_string(reply, "result", "ok")
+ xpc_connection_send_message(connection, reply)
+ default:
+ self.lock.withLock {
+ self.pendingRequests.append(object)
+ }
+ }
+ case XPC_TYPE_ERROR:
+ break
+ default:
+ fatalError("unhandled xpc object type: \(xpc_get_type(object))")
+ }
+ }
+ xpc_connection_activate(connection)
+ }
+}
+#endif