diff --git a/BitMatch/Core/Services/SFTPRemoteBackupProvider.swift b/BitMatch/Core/Services/SFTPRemoteBackupProvider.swift index f1fc0c2b..32cb033f 100644 --- a/BitMatch/Core/Services/SFTPRemoteBackupProvider.swift +++ b/BitMatch/Core/Services/SFTPRemoteBackupProvider.swift @@ -20,6 +20,8 @@ enum SFTPRemoteBackupProviderFactory { } #if os(macOS) +import Darwin + struct OpenSSHCommand: Equatable, Sendable { let executable: String let arguments: [String] @@ -41,14 +43,111 @@ private final class OpenSSHProcessOutput: @unchecked Sendable { private let lock = NSLock() private var stdout = Data() private var stderr = Data() + private var inputWriteFailed = false func setStdout(_ data: Data) { lock.withLock { stdout = data } } func setStderr(_ data: Data) { lock.withLock { stderr = data } } + func markInputWriteFailure() { lock.withLock { inputWriteFailed = true } } + func didInputWriteFail() -> Bool { lock.withLock { inputWriteFailed } } func result(status: Int32) -> OpenSSHCommandResult { lock.withLock { OpenSSHCommandResult(status: status, stdout: stdout, stderr: stderr) } } } +/// Owns one launched OpenSSH command until it exits and both pipes drain. A +/// ProxyCommand may leave descendants holding those pipes after ssh exits, so +/// cancellation signals the group created atomically with the child. +private final class OpenSSHProcessLifetime: @unchecked Sendable { + private let lock = NSLock() + private var childPID: pid_t? + private var cancelled = false + private var finished = false + private var forceKill: DispatchWorkItem? + + func launch(_ command: OpenSSHCommand, input: Pipe, output: Pipe, error: Pipe) throws -> pid_t { + try lock.withLock { + guard !cancelled else { throw CancellationError() } + var actions: posix_spawn_file_actions_t? + try Self.check(posix_spawn_file_actions_init(&actions)) + defer { posix_spawn_file_actions_destroy(&actions) } + + let inputRead = input.fileHandleForReading.fileDescriptor + let outputWrite = output.fileHandleForWriting.fileDescriptor + let errorWrite = error.fileHandleForWriting.fileDescriptor + try Self.check(posix_spawn_file_actions_adddup2(&actions, inputRead, STDIN_FILENO)) + try Self.check(posix_spawn_file_actions_adddup2(&actions, outputWrite, STDOUT_FILENO)) + try Self.check(posix_spawn_file_actions_adddup2(&actions, errorWrite, STDERR_FILENO)) + for descriptor in [inputRead, outputWrite, errorWrite, + input.fileHandleForWriting.fileDescriptor, + output.fileHandleForReading.fileDescriptor, + error.fileHandleForReading.fileDescriptor] where descriptor > STDERR_FILENO { + try Self.check(posix_spawn_file_actions_addclose(&actions, descriptor)) + } + + var attributes: posix_spawnattr_t? + try Self.check(posix_spawnattr_init(&attributes)) + defer { posix_spawnattr_destroy(&attributes) } + try Self.check(posix_spawnattr_setflags( + &attributes, Int16(POSIX_SPAWN_SETPGROUP | POSIX_SPAWN_CLOEXEC_DEFAULT) + )) + try Self.check(posix_spawnattr_setpgroup(&attributes, 0)) + + let strings = ([command.executable] + command.arguments).map { $0.withCString { strdup($0) } } + defer { strings.forEach { free($0) } } + guard strings.allSatisfy({ $0 != nil }) else { throw POSIXError(.ENOMEM) } + var arguments = strings + [nil] + var pid: pid_t = 0 + let spawnError = command.executable.withCString { executable in + arguments.withUnsafeMutableBufferPointer { buffer in + posix_spawn(&pid, executable, &actions, &attributes, buffer.baseAddress, environ) + } + } + try Self.check(spawnError) + childPID = pid + return pid + } + } + + private static func check(_ code: Int32) throws { + if code != 0 { throw NSError(domain: NSPOSIXErrorDomain, code: Int(code)) } + } + + func cancel() { + lock.withLock { + guard !finished else { return } + cancelled = true + guard childPID != nil else { return } + signal(SIGTERM) + let work = DispatchWorkItem { [weak self] in self?.forceKillIfUnfinished() } + forceKill = work + DispatchQueue.global().asyncAfter(deadline: .now() + 2, execute: work) + } + } + + private func forceKillIfUnfinished() { + lock.withLock { + guard !finished else { return } + signal(SIGKILL) + } + } + + /// A terminated direct child does not mean its pipe-owning descendants + /// have exited. Keep signalling the owned group until both drains finish. + private func signal(_ value: Int32) { + if let childPID, childPID > 0 { _ = Darwin.kill(-childPID, value) } + } + + func complete() -> Bool { + lock.withLock { + finished = true + childPID = nil + forceKill?.cancel() + forceKill = nil + return cancelled + } + } +} + actor SFTPRemoteBackupProvider: RemoteBackupProvider { private let profile: RemoteDestinationProfile private let credential: RemoteCredential @@ -284,43 +383,100 @@ actor SFTPRemoteBackupProvider: RemoteBackupProvider { : .capabilityUnavailable(.noReplacePromotion) } - private static func runProcess(_ command: OpenSSHCommand) async throws -> OpenSSHCommandResult { - try await withCheckedThrowingContinuation { continuation in - let process = Process() - let output = Pipe(); let error = Pipe(); let input = Pipe() - let collected = OpenSSHProcessOutput() - let drains = DispatchGroup() - process.executableURL = URL(fileURLWithPath: command.executable) - process.arguments = command.arguments - process.standardOutput = output; process.standardError = error; process.standardInput = input - // Reserve both drain completions before the child can terminate. - // Otherwise a fast child could notify an empty group before these - // readers have been registered. - drains.enter() - drains.enter() - process.terminationHandler = { completed in - drains.notify(queue: .global()) { - continuation.resume(returning: collected.result(status: completed.terminationStatus)) - } - } - do { - try process.run() - DispatchQueue.global().async { - collected.setStdout(output.fileHandleForReading.readDataToEndOfFile()) + // Kept internal so the real subprocess lifetime can be exercised by tests. + static func runProcess(_ command: OpenSSHCommand) async throws -> OpenSSHCommandResult { + let lifetime = OpenSSHProcessLifetime() + let result: OpenSSHCommandResult = try await withTaskCancellationHandler(operation: { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + let output = Pipe(); let stderrPipe = Pipe(); let input = Pipe() + let collected = OpenSSHProcessOutput() + let drains = DispatchGroup() + drains.enter() + drains.enter() + drains.enter() + do { + // A fast-exiting ssh/sftp child can close stdin while a + // batch is being written. Return EPIPE, never SIGPIPE. + guard Darwin.fcntl(input.fileHandleForWriting.fileDescriptor, F_SETNOSIGPIPE, 1) != -1 else { + throw NSError(domain: NSPOSIXErrorDomain, code: Int(errno)) + } + let pid = try lifetime.launch(command, input: input, output: output, error: stderrPipe) + // Only the child keeps the duplicated standard streams. + input.fileHandleForReading.closeFile() + output.fileHandleForWriting.closeFile() + stderrPipe.fileHandleForWriting.closeFile() + DispatchQueue.global().async { + collected.setStdout(output.fileHandleForReading.readDataToEndOfFile()) + output.fileHandleForReading.closeFile() + drains.leave() + } + DispatchQueue.global().async { + collected.setStderr(stderrPipe.fileHandleForReading.readDataToEndOfFile()) + stderrPipe.fileHandleForReading.closeFile() + drains.leave() + } + // Do not reap the group leader until its output closes. + // Observe its exit without reaping, then disarm the group + // timer while its PID is still reserved. Only then reap. + drains.notify(queue: .global()) { + var exitInfo = siginfo_t() + var observed: Int32 + repeat { + observed = Darwin.waitid(P_PID, id_t(pid), &exitInfo, WEXITED | WNOWAIT) + } while observed == -1 && errno == EINTR + let wasCancelled = lifetime.complete() + var rawStatus: Int32 = 0 + var waited: pid_t + repeat { waited = Darwin.waitpid(pid, &rawStatus, 0) } + while waited == -1 && errno == EINTR + let status = observed == 0 && exitInfo.si_pid == pid && waited == pid + ? Self.exitStatus(rawStatus) : nil + if wasCancelled { + continuation.resume(throwing: CancellationError()) + } else if let status { + if status == 0 && collected.didInputWriteFail() { + continuation.resume(throwing: RemoteBackupError.providerUnavailable) + } else { + continuation.resume(returning: collected.result(status: status)) + } + } else { + continuation.resume(throwing: RemoteBackupError.providerUnavailable) + } + } + if let data = command.standardInput { + do { try input.fileHandleForWriting.write(contentsOf: data) } + catch { collected.markInputWriteFailure() } + } + input.fileHandleForWriting.closeFile() + drains.leave() + } catch { + input.fileHandleForReading.closeFile() + input.fileHandleForWriting.closeFile() + output.fileHandleForReading.closeFile() + output.fileHandleForWriting.closeFile() + stderrPipe.fileHandleForReading.closeFile() + stderrPipe.fileHandleForWriting.closeFile() + let wasCancelled = lifetime.complete() + drains.leave() drains.leave() - } - DispatchQueue.global().async { - collected.setStderr(error.fileHandleForReading.readDataToEndOfFile()) drains.leave() + if wasCancelled { + continuation.resume(throwing: CancellationError()) + } else { + continuation.resume(throwing: error) + } } - if let data = command.standardInput { input.fileHandleForWriting.write(data) } - input.fileHandleForWriting.closeFile() - } catch { - drains.leave() - drains.leave() - continuation.resume(throwing: error) } - } + }, onCancel: { + lifetime.cancel() + }) + try Task.checkCancellation() + return result + } + + private nonisolated static func exitStatus(_ rawStatus: Int32) -> Int32 { + let signal = rawStatus & 0x7f + return signal == 0 ? (rawStatus >> 8) & 0xff : 128 + signal } } #endif diff --git a/BitMatchTests/SFTPRemoteBackupProcessTests.swift b/BitMatchTests/SFTPRemoteBackupProcessTests.swift new file mode 100644 index 00000000..22aa291a --- /dev/null +++ b/BitMatchTests/SFTPRemoteBackupProcessTests.swift @@ -0,0 +1,187 @@ +import Darwin +import Foundation +import Testing +import BitMatchEngine +@testable import BitMatch + +#if os(macOS) +struct SFTPRemoteBackupProcessTests { + @Test func completedCommandStillReturnsStatusAndBothOutputStreams() async throws { + let result = try await SFTPRemoteBackupProvider.runProcess(OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "printf output; printf warning >&2; exit 7"], + standardInput: nil + )) + + #expect(result.status == 7) + #expect(String(decoding: result.stdout, as: UTF8.self) == "output") + #expect(String(decoding: result.stderr, as: UTF8.self) == "warning") + } + + @Test func completedCommandReceivesBatchInput() async throws { + let batch = Data("put one\nput two\n".utf8) + let result = try await SFTPRemoteBackupProvider.runProcess(OpenSSHCommand( + executable: "/bin/cat", arguments: [], standardInput: batch + )) + + #expect(result.status == 0) + #expect(result.stdout == batch) + #expect(result.stderr.isEmpty) + } + + @Test func earlyExitWhileWritingInputCannotReportSuccess() async throws { + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "exec 0<&-; exit 0"], + standardInput: Data(repeating: 0x61, count: 4 * 1024 * 1024) + ) + + await #expect(throws: RemoteBackupError.self) { + try await SFTPRemoteBackupProvider.runProcess(command) + } + } + + @Test func cancellingInFlightCommandStopsAndReapsChild() async throws { + let marker = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-child-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: marker) } + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "printf '%s\\n' \"$$\" > \"$1\"; exec /bin/sleep 5", "sh", marker.path], + standardInput: nil + ) + let task = Task { try await SFTPRemoteBackupProvider.runProcess(command) } + guard await waitUntil(timeout: .seconds(3), { FileManager.default.fileExists(atPath: marker.path) }) else { + task.cancel() + _ = try? await task.value + Issue.record("The child process never started") + return + } + let pid = try #require(Int32(String(contentsOf: marker, encoding: .utf8).trimmingCharacters(in: .whitespacesAndNewlines))) + let started = ContinuousClock.now + task.cancel() + + await #expect(throws: CancellationError.self) { try await task.value } + #expect(ContinuousClock.now - started < .seconds(3)) + #expect(Darwin.kill(pid, 0) == -1 && errno == ESRCH) + } + + @Test func alreadyCancelledCommandNeverStartsAChild() async throws { + let marker = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-child-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: marker) } + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "printf started > \"$1\"", "sh", marker.path], + standardInput: nil + ) + let task = Task { + withUnsafeCurrentTask { $0?.cancel() } + return try await SFTPRemoteBackupProvider.runProcess(command) + } + + await #expect(throws: CancellationError.self) { try await task.value } + #expect(!FileManager.default.fileExists(atPath: marker.path)) + } + + @Test func cancellationEscalatesWhenChildIgnoresTerminate() async throws { + let marker = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-child-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: marker) } + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "trap '' TERM; printf '%s\\n' \"$$\" > \"$1\"; exec /bin/sleep 8", "sh", marker.path], + standardInput: nil + ) + let task = Task { try await SFTPRemoteBackupProvider.runProcess(command) } + guard await waitUntil(timeout: .seconds(3), { FileManager.default.fileExists(atPath: marker.path) }) else { + task.cancel() + _ = try? await task.value + Issue.record("The TERM-ignoring child process never started") + return + } + let pid = try #require(Int32(String(contentsOf: marker, encoding: .utf8).trimmingCharacters(in: .whitespacesAndNewlines))) + let started = ContinuousClock.now + task.cancel() + + await #expect(throws: CancellationError.self) { try await task.value } + #expect(ContinuousClock.now - started < .seconds(5)) + #expect(Darwin.kill(pid, 0) == -1 && errno == ESRCH) + } + + @Test func cancellationStillStopsChildThatClosedBothOutputPipes() async throws { + let marker = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-no-output-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: marker) } + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", "exec 1>&- 2>&-; printf '%s\\n' \"$$\" > \"$1\"; exec /bin/sleep 8", "sh", marker.path], + standardInput: nil + ) + let task = Task { try await SFTPRemoteBackupProvider.runProcess(command) } + guard await waitUntil(timeout: .seconds(3), { FileManager.default.fileExists(atPath: marker.path) }) else { + task.cancel() + _ = try? await task.value + Issue.record("The output-closing child never started") + return + } + let pid = try #require(Int32(String(contentsOf: marker, encoding: .utf8).trimmingCharacters(in: .whitespacesAndNewlines))) + let started = ContinuousClock.now + task.cancel() + + await #expect(throws: CancellationError.self) { try await task.value } + #expect(ContinuousClock.now - started < .seconds(3)) + #expect(Darwin.kill(pid, 0) == -1 && errno == ESRCH) + } + + @Test func cancellationStopsPipeOwningDescendantAfterDirectChildExits() async throws { + let marker = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-family-\(UUID().uuidString)") + let ready = FileManager.default.temporaryDirectory.appendingPathComponent("bitmatch-sftp-ready-\(UUID().uuidString)") + defer { + try? FileManager.default.removeItem(at: marker) + try? FileManager.default.removeItem(at: ready) + } + let script = #""" + /bin/sh -c 'trap "" TERM; printf ready > "$1"; exec /bin/sleep 8' sh "$2" & + child=$! + printf '%s\n%s\n' "$$" "$child" > "$1" + exit 0 + """# + let command = OpenSSHCommand( + executable: "/bin/sh", + arguments: ["-c", script, "sh", marker.path, ready.path], + standardInput: nil + ) + let task = Task { try await SFTPRemoteBackupProvider.runProcess(command) } + guard await waitUntil(timeout: .seconds(3), { + FileManager.default.fileExists(atPath: marker.path) && FileManager.default.fileExists(atPath: ready.path) + }) else { + task.cancel() + _ = try? await task.value + Issue.record("The pipe-owning descendant never started") + return + } + let pids = try String(contentsOf: marker, encoding: .utf8) + .split(whereSeparator: \.isNewline) + .compactMap { Int32($0) } + #expect(pids.count == 2) + let parentPID = try #require(pids.first) + let childPID = try #require(pids.last) + guard await waitUntil(timeout: .seconds(3), { + var exitInfo = siginfo_t() + return Darwin.waitid(P_PID, id_t(parentPID), &exitInfo, + WEXITED | WNOWAIT | WNOHANG) == 0 + && exitInfo.si_pid == parentPID + }) else { + task.cancel() + _ = try? await task.value + Issue.record("The direct child did not exit before cancellation") + return + } + let started = ContinuousClock.now + task.cancel() + + await #expect(throws: CancellationError.self) { try await task.value } + #expect(ContinuousClock.now - started < .seconds(5)) + #expect(await waitUntil(timeout: .seconds(2), { + Darwin.kill(childPID, 0) == -1 && errno == ESRCH + })) + } +} +#endif