diff --git a/Package.swift b/Package.swift index 34d18c7d7..2da0922a2 100644 --- a/Package.swift +++ b/Package.swift @@ -150,6 +150,7 @@ let package = Package( .testTarget( name: "ContainerCommandsTests", dependencies: [ + "ContainerAPIClient", "ContainerCommands", "ContainerResource", ] @@ -474,6 +475,12 @@ let package = Package( "CAuditToken", ] ), + .testTarget( + name: "ContainerXPCTests", + dependencies: [ + "ContainerXPC" + ] + ), .target( name: "ContainerOS", dependencies: [ diff --git a/Sources/ContainerCommands/Application.swift b/Sources/ContainerCommands/Application.swift index c77f51a59..9756d0edc 100644 --- a/Sources/ContainerCommands/Application.swift +++ b/Sources/ContainerCommands/Application.swift @@ -36,6 +36,8 @@ private nonisolated(unsafe) var bootstrapLogger = { return log }() +typealias APIServerHealthCheck = @Sendable (Duration) async throws -> SystemHealth + public struct Application: AsyncLoggableCommand { @OptionGroup public var logOptions: Flags.Logging @@ -127,9 +129,8 @@ public struct Application: AsyncLoggableCommand { } catch { // --help/-h on the root command (e.g. `container --help`) is intercepted // by ArgumentParser and lands here. - let containsHelp = fullArgs.contains("-h") || fullArgs.contains("--help") - if fullArgs.count <= 2 && containsHelp { - let pluginLoader = try? await createPluginLoader() + if isRootHelpRequest(fullArgs) { + let pluginLoader = await pluginLoaderForHelp() await Self.printModifiedHelpText(pluginLoader: pluginLoader) return } @@ -144,6 +145,18 @@ public struct Application: AsyncLoggableCommand { } public static func createPluginLoader() async throws -> PluginLoader { + try await createPluginLoader(healthTimeout: .seconds(10), wallTimeout: .seconds(10), healthCheck: defaultAPIServerHealthCheck(timeout:)) + } + + static func pluginLoaderForHelp(healthCheck: @escaping APIServerHealthCheck = defaultAPIServerHealthCheck(timeout:)) async -> PluginLoader? { + try? await createPluginLoader(healthTimeout: .seconds(1), wallTimeout: .seconds(1), healthCheck: healthCheck) + } + + static func isRootHelpRequest(_ arguments: [String]) -> Bool { + arguments.count <= 2 && (arguments.contains("-h") || arguments.contains("--help")) + } + + private static func createPluginLoader(healthTimeout: Duration, wallTimeout: Duration, healthCheck: @escaping APIServerHealthCheck) async throws -> PluginLoader { let installRootPath = CommandLine.executablePath .removingLastComponent() .removingLastComponent() @@ -175,9 +188,7 @@ public struct Application: AsyncLoggableCommand { AppBundlePluginFactory(logger: bootstrapLogger), ] - guard let systemHealth = try? await ClientHealthCheck.ping(timeout: .seconds(10)) else { - throw ContainerizationError(.timeout, message: "unable to retrieve application data root from API server") - } + let systemHealth = try await apiServerHealth(healthTimeout: healthTimeout, wallTimeout: wallTimeout, healthCheck: healthCheck) return try PluginLoader( appRoot: systemHealth.appRoot, installRoot: systemHealth.installRoot, @@ -188,6 +199,43 @@ public struct Application: AsyncLoggableCommand { ) } + private static func defaultAPIServerHealthCheck(timeout: Duration) async throws -> SystemHealth { + try await ClientHealthCheck.ping(timeout: timeout) + } + + static func apiServerHealth( + healthTimeout: Duration, + wallTimeout: Duration, + healthCheck: @escaping APIServerHealthCheck = defaultAPIServerHealthCheck(timeout:) + ) async throws -> SystemHealth { + try await withCheckedThrowingContinuation { continuation in + let result = HealthCheckResult(continuation) + + let healthTask = Task.detached { + do { + let health = try await healthCheck(healthTimeout) + await result.resume(.success(health)) + } catch { + await result.resume(.failure(error)) + } + } + + let timeoutTask = Task.detached { + try? await Task.sleep(for: wallTimeout) + await result.resume( + .failure( + ContainerizationError( + .timeout, + message: "unable to retrieve application data root from API server" + ))) + } + + Task { + await result.setTasks(healthTask: healthTask, timeoutTask: timeoutTask) + } + } + } + /// Load the system configuration using `appRoot` / `installRoot` reported by the /// daemon. `container system start` MUST have previously been run to start the daemon. public static func loadContainerSystemConfig() async throws -> ContainerSystemConfig { @@ -254,6 +302,41 @@ public struct Application: AsyncLoggableCommand { } } +private actor HealthCheckResult { + private let continuation: CheckedContinuation + private var healthTask: Task? + private var timeoutTask: Task? + private var resumed = false + + init(_ continuation: CheckedContinuation) { + self.continuation = continuation + } + + func setTasks(healthTask: Task, timeoutTask: Task) { + if resumed { + healthTask.cancel() + timeoutTask.cancel() + return + } + self.healthTask = healthTask + self.timeoutTask = timeoutTask + } + + func resume(_ result: Result) { + guard !resumed else { + return + } + resumed = true + let healthTask = healthTask + let timeoutTask = timeoutTask + self.healthTask = nil + self.timeoutTask = nil + healthTask?.cancel() + timeoutTask?.cancel() + continuation.resume(with: result) + } +} + extension Application { // Because we support plugins, we need to modify the help text to display // any if we found some. diff --git a/Sources/ContainerCommands/DefaultCommand.swift b/Sources/ContainerCommands/DefaultCommand.swift index 3df8f570e..e79572542 100644 --- a/Sources/ContainerCommands/DefaultCommand.swift +++ b/Sources/ContainerCommands/DefaultCommand.swift @@ -34,9 +34,8 @@ struct DefaultCommand: AsyncLoggableCommand { var remaining: [String] = [] func run() async throws { - // See if we have a possible plugin command. - let pluginLoader = try? await Application.createPluginLoader() guard let command = remaining.first else { + let pluginLoader = await Application.pluginLoaderForHelp() await Application.printModifiedHelpText(pluginLoader: pluginLoader) return } @@ -66,6 +65,9 @@ struct DefaultCommand: AsyncLoggableCommand { .map { $0.appendingPathComponent(command).path(percentEncoded: false) } .joined(separator: "\n - ") + // See if we have a possible plugin command. + let pluginLoader = try? await Application.createPluginLoader() + // If plugin loader couldn't be created, the system/APIServer likely isn't running. if pluginLoader == nil { throw ValidationError( diff --git a/Sources/ContainerCommands/HelpCommand.swift b/Sources/ContainerCommands/HelpCommand.swift index 43e9107c5..cf6458402 100644 --- a/Sources/ContainerCommands/HelpCommand.swift +++ b/Sources/ContainerCommands/HelpCommand.swift @@ -31,7 +31,7 @@ struct HelpCommand: AsyncLoggableCommand { func run() async throws { if subcommandPath.isEmpty { - let pluginLoader = try? await Application.createPluginLoader() + let pluginLoader = await Application.pluginLoaderForHelp() await Application.printModifiedHelpText(pluginLoader: pluginLoader) return } diff --git a/Sources/ContainerXPC/XPCClient.swift b/Sources/ContainerXPC/XPCClient.swift index bab008509..d8c63c49d 100644 --- a/Sources/ContainerXPC/XPCClient.swift +++ b/Sources/ContainerXPC/XPCClient.swift @@ -98,42 +98,31 @@ extension XPCClient { /// Send the provided message to the service. @discardableResult public func send(_ message: XPCMessage, responseTimeout: Duration? = nil) async throws -> XPCMessage { - try await withThrowingTaskGroup(of: XPCMessage.self, returning: XPCMessage.self) { group in + try await withCheckedThrowingContinuation { continuation in + let result = XPCResponseResult(continuation) + 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 route = message.string(key: XPCMessage.routeKey) ?? "nil" + let timeoutTask = Task.detached { + try? await Task.sleep(for: responseTimeout) + _ = result.resume( + .failure( + ContainerizationError( + .timeout, + message: "XPC timeout for request to \(self.service)/\(route)" + ))) } + result.setTimeoutTask(timeoutTask) } - 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) - } - } + xpc_connection_send_message_with_reply(self.connection, message.underlying, nil) { reply in + do { + let message = try self.parseReply(reply) + _ = result.resume(.success(message)) + } catch { + _ = result.resume(.failure(error)) } } - - 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") - } - return response } } @@ -159,4 +148,42 @@ extension XPCClient { } } +private final class XPCResponseResult: @unchecked Sendable { + private let lock = NSLock() + private let continuation: CheckedContinuation + private var timeoutTask: Task? + private var resumed = false + + init(_ continuation: CheckedContinuation) { + self.continuation = continuation + } + + func setTimeoutTask(_ task: Task) { + lock.lock() + if resumed { + lock.unlock() + task.cancel() + return + } + timeoutTask = task + lock.unlock() + } + + func resume(_ result: Result) -> Bool { + lock.lock() + guard !resumed else { + lock.unlock() + return false + } + resumed = true + let timeoutTask = timeoutTask + self.timeoutTask = nil + lock.unlock() + + timeoutTask?.cancel() + continuation.resume(with: result) + return true + } +} + #endif diff --git a/Tests/ContainerCommandsTests/ApplicationHealthTests.swift b/Tests/ContainerCommandsTests/ApplicationHealthTests.swift new file mode 100644 index 000000000..52504bb2b --- /dev/null +++ b/Tests/ContainerCommandsTests/ApplicationHealthTests.swift @@ -0,0 +1,276 @@ +//===----------------------------------------------------------------------===// +// 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. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerizationError +import Darwin +import Foundation +import Testing + +@testable import ContainerCommands + +@Suite(.serialized) +struct ApplicationHealthTests { + @Test(arguments: [ + (["container", "--help"], true), + (["container", "-h"], true), + (["container"], false), + (["container", "compose", "--help"], false), + (["container", "--version"], false), + ]) + func rootHelpRequestDetection(arguments: [String], expected: Bool) { + #expect(Application.isRootHelpRequest(arguments) == expected) + } + + @Test + func pluginLoaderForHelpReturnsLoaderWhenHealthCheckSucceeds() async throws { + let appRoot = try Self.makeTemporaryDirectory() + defer { try? FileManager.default.removeItem(at: appRoot) } + + let loader = await Application.pluginLoaderForHelp { timeout in + #expect(timeout == .seconds(1)) + return try Self.makeSystemHealth(appRoot: appRoot) + } + + #expect(loader != nil) + } + + @Test + func pluginLoaderForHelpReturnsNilWhenHealthCheckTimesOut() async throws { + let loader = await Application.pluginLoaderForHelp { _ in + try await Task.sleep(for: .seconds(2)) + return try Self.makeSystemHealth() + } + + #expect(loader == nil) + } + + @Test + func defaultCommandWithoutArgumentsCompletesWhenPluginDiscoveryIsUnavailable() async throws { + var command = DefaultCommand() + command.remaining = [] + let box = UncheckedSendableBox(command) + + try await Self.requireCompletesWithin(.seconds(2)) { + try await Self.discardStandardOutput { + try await box.value.run() + } + } + } + + @Test + func helpCommandWithoutPathCompletesWhenPluginDiscoveryIsUnavailable() async throws { + var command = HelpCommand() + command.subcommandPath = [] + let box = UncheckedSendableBox(command) + + try await Self.requireCompletesWithin(.seconds(2)) { + try await Self.discardStandardOutput { + try await box.value.run() + } + } + } + + @Test + func apiServerHealthReturnsSuccessfulHealthCheck() async throws { + let expected = try Self.makeSystemHealth() + + let health = try await Application.apiServerHealth( + healthTimeout: .seconds(10), + wallTimeout: .seconds(1) + ) { timeout in + #expect(timeout == .seconds(10)) + return expected + } + + #expect(health.appRoot == expected.appRoot) + #expect(health.installRoot == expected.installRoot) + #expect(health.logRoot == expected.logRoot) + #expect(health.apiServerCommit == expected.apiServerCommit) + } + + @Test + func apiServerHealthReturnsHealthCheckErrorBeforeWallTimeout() async throws { + do { + _ = try await Application.apiServerHealth( + healthTimeout: .seconds(10), + wallTimeout: .seconds(1) + ) { _ in + throw ContainerizationError(.invalidState, message: "health check failed") + } + Issue.record("expected health check to throw") + } catch let error as ContainerizationError { + #expect(error.code == .invalidState) + #expect(error.message == "health check failed") + } + } + + @Test + func apiServerHealthReturnsWallTimeoutWhenHealthCheckDoesNotComplete() async throws { + let clock = ContinuousClock() + let start = clock.now + + do { + _ = try await Application.apiServerHealth( + healthTimeout: .seconds(10), + wallTimeout: .milliseconds(100) + ) { _ in + try await Task.sleep(for: .seconds(1)) + return try Self.makeSystemHealth() + } + Issue.record("expected health check to time out") + } catch let error as ContainerizationError { + let elapsed = start.duration(to: clock.now) + #expect(error.code == .timeout) + #expect(error.message == "unable to retrieve application data root from API server") + #expect(elapsed < .seconds(2)) + } + } + + @Test + func apiServerHealthIgnoresLateHealthCheckAfterWallTimeout() async throws { + do { + _ = try await Application.apiServerHealth( + healthTimeout: .seconds(10), + wallTimeout: .milliseconds(50) + ) { _ in + try await Task.sleep(for: .milliseconds(150)) + return try Self.makeSystemHealth() + } + Issue.record("expected health check to time out") + } catch let error as ContainerizationError { + #expect(error.code == .timeout) + } + + try await Task.sleep(for: .milliseconds(250)) + } + + private static func makeSystemHealth(appRoot: URL = URL(filePath: "/tmp/container-test-app-root")) throws -> SystemHealth { + let json = """ + { + "appRoot": "\(appRoot.absoluteString)", + "installRoot": "file:///tmp/container-test-install-root", + "apiServerVersion": "test-version", + "apiServerCommit": "test-commit", + "apiServerBuild": "debug", + "apiServerAppName": "container-apiserver" + } + """ + return try JSONDecoder().decode(SystemHealth.self, from: Data(json.utf8)) + } + + private static func makeTemporaryDirectory() throws -> URL { + let url = FileManager.default.temporaryDirectory.appending(path: UUID().uuidString) + try FileManager.default.createDirectory(at: url, withIntermediateDirectories: true) + return url + } + + private static func requireCompletesWithin(_ timeout: Duration, operation: @escaping @Sendable () async throws -> Void) async throws { + try await withCheckedThrowingContinuation { continuation in + let result = VoidAsyncResult(continuation) + + let operationTask = Task { + do { + try await operation() + await result.resume(.success(())) + } catch { + await result.resume(.failure(error)) + } + } + + let timeoutTask = Task { + try? await Task.sleep(for: timeout) + await result.resume( + .failure( + ContainerizationError( + .timeout, + message: "operation did not complete within \(timeout)" + ))) + } + + Task { + await result.setTasks(operationTask: operationTask, timeoutTask: timeoutTask) + } + } + } + + private static func discardStandardOutput(operation: () async throws -> Void) async throws { + fflush(stdout) + let original = dup(STDOUT_FILENO) + let pipe = Pipe() + var restored = false + dup2(pipe.fileHandleForWriting.fileDescriptor, STDOUT_FILENO) + defer { + fflush(stdout) + if !restored { + dup2(original, STDOUT_FILENO) + pipe.fileHandleForWriting.closeFile() + } + close(original) + pipe.fileHandleForReading.closeFile() + } + + try await operation() + fflush(stdout) + dup2(original, STDOUT_FILENO) + restored = true + pipe.fileHandleForWriting.closeFile() + pipe.fileHandleForReading.readDataToEndOfFile() + } +} + +private actor VoidAsyncResult { + private let continuation: CheckedContinuation + private var operationTask: Task? + private var timeoutTask: Task? + private var resumed = false + + init(_ continuation: CheckedContinuation) { + self.continuation = continuation + } + + func setTasks(operationTask: Task, timeoutTask: Task) { + if resumed { + operationTask.cancel() + timeoutTask.cancel() + return + } + self.operationTask = operationTask + self.timeoutTask = timeoutTask + } + + func resume(_ result: Result) { + guard !resumed else { + return + } + resumed = true + let operationTask = operationTask + let timeoutTask = timeoutTask + self.operationTask = nil + self.timeoutTask = nil + operationTask?.cancel() + timeoutTask?.cancel() + continuation.resume(with: result) + } +} + +private final class UncheckedSendableBox: @unchecked Sendable { + let value: Value + + init(_ value: Value) { + self.value = value + } +} diff --git a/Tests/ContainerXPCTests/XPCClientTests.swift b/Tests/ContainerXPCTests/XPCClientTests.swift new file mode 100644 index 000000000..5d51b4d6b --- /dev/null +++ b/Tests/ContainerXPCTests/XPCClientTests.swift @@ -0,0 +1,165 @@ +//===----------------------------------------------------------------------===// +// 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 ContainerXPC +import ContainerizationError +@preconcurrency import Foundation +import Testing + +@Suite(.timeLimit(.minutes(1)), .serialized) +struct XPCClientTests { + @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 { + let elapsed = start.duration(to: clock.now) + #expect(error.code == .timeout) + #expect(error.message.contains("XPC timeout for request to test.container.xpc/hang")) + #expect(elapsed < .seconds(2)) + } + } + + @Test + func responseTimeoutHandlesMessagesWithoutRoute() async throws { + let server = AnonymousXPCServer() + defer { server.close() } + + let client = server.makeClient() + + do { + _ = try await client.send(XPCMessage(object: xpc_dictionary_create_empty()), responseTimeout: .milliseconds(100)) + Issue.record("expected send to time out") + } catch let error as ContainerizationError { + #expect(error.code == .timeout) + #expect(error.message.contains("XPC timeout for request to test.container.xpc/nil")) + } + } + + @Test + func sendWithoutResponseTimeoutCompletesWhenServerReplies() async throws { + let server = AnonymousXPCServer() + defer { server.close() } + + let client = server.makeClient() + let response = try await client.send(XPCMessage(route: "echo")) + #expect(response.string(key: "result") == "ok") + } + + @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.code == .timeout) + } + + 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() + } + } + + 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) + case "hang": + self.lock.withLock { + self.pendingRequests.append(object) + } + 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