From 255515ddb1e06bd2768f8fa780ac676bfe183b89 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Sat, 26 Sep 2026 03:15:53 +0700 Subject: [PATCH] fix(mcp): keep tablepro-mcp reading a non-blocking stdin and drain every pipe without O_NONBLOCK --- CHANGELOG.md | 1 + TablePro/CLI/BridgeProxy.swift | 2 +- .../Core/Database/CLIToolVersionProbe.swift | 26 +--------- .../Database/ProcessNativeDumpRunner.swift | 16 +++--- TablePro/Core/Process/DescriptorRead.swift | 52 +++++++++++++------ TablePro/Core/Process/PipeReader.swift | 25 ++------- .../Process/SupervisedProcessRunner.swift | 5 -- .../Core/MCP/Transport/BridgeStdinTests.swift | 50 ++++++++++++++++++ .../Core/Process/DescriptorReadTests.swift | 51 +++++++++++++++--- .../Core/Process/PipeReaderTests.swift | 40 +++++++------- .../PreConnectHookRunnerTests.swift | 4 -- .../ProcessNativeDumpRunnerTests.swift | 39 ++++++++++++++ .../Helpers/BackgroundPipeWriter.swift | 20 +++++++ TableProTests/Helpers/HeldOpenWriter.swift | 46 ++++++++++++++++ 14 files changed, 266 insertions(+), 111 deletions(-) create mode 100644 TableProTests/Core/MCP/Transport/BridgeStdinTests.swift create mode 100644 TableProTests/Helpers/BackgroundPipeWriter.swift create mode 100644 TableProTests/Helpers/HeldOpenWriter.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 0767f4655b..7288ba69e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -128,6 +128,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Pre-connect script failures sometimes reported without the script's own error message. +- `tablepro-mcp` crashing when its standard input was non-blocking. - Server connections piling up while browsing many databases or schemas, and staying open after a failed connect. (#3103) - Variables declared in a SQL Server script lost after its first statement. (#3078) - Later SQL Server result sets shown under the first one's columns, or crashing the app. diff --git a/TablePro/CLI/BridgeProxy.swift b/TablePro/CLI/BridgeProxy.swift index 0c5d5cb056..4884d949f1 100644 --- a/TablePro/CLI/BridgeProxy.swift +++ b/TablePro/CLI/BridgeProxy.swift @@ -532,7 +532,7 @@ enum BridgeStdin { while !Task.isCancelled { let chunk: Data do { - chunk = try DescriptorRead.availableBytes(from: descriptor) + chunk = try DescriptorRead.nextBytes(from: descriptor) } catch { logger.log(.error, "Reading stdin failed: \(error.localizedDescription)") break diff --git a/TablePro/Core/Database/CLIToolVersionProbe.swift b/TablePro/Core/Database/CLIToolVersionProbe.swift index 2faec18143..6cc5055df5 100644 --- a/TablePro/Core/Database/CLIToolVersionProbe.swift +++ b/TablePro/Core/Database/CLIToolVersionProbe.swift @@ -46,29 +46,7 @@ enum CLIToolVersionProbe { return nil } guard process.terminationStatus == 0 else { return nil } - return String(data: readAvailable(from: pipe.fileHandleForReading), encoding: .utf8) - } - - /// What the pipe holds now, rather than what it holds at EOF. - /// - /// The tool has exited, so its own output is already here. `readDataToEndOfFile` would wait for - /// every writer to close instead, and a wrapper that prints its version, starts a helper that - /// inherits standard output and exits leaves that EOF to the helper: the deadline above covers - /// only the process, so the read has none and the dump never starts. - private static func readAvailable(from handle: FileHandle) -> Data { - let descriptor = handle.fileDescriptor - let flags = fcntl(descriptor, F_GETFL) - guard flags != -1, fcntl(descriptor, F_SETFL, flags | O_NONBLOCK) != -1 else { return Data() } - - var output = Data() - var buffer = [UInt8](repeating: 0, count: 4_096) - while output.count < outputCap { - let received = buffer.withUnsafeMutableBytes { raw in - read(descriptor, raw.baseAddress, raw.count) - } - guard received > 0 else { break } - output.append(contentsOf: buffer[0 ..< received]) - } - return output + let output = DescriptorRead.bufferedBytes(from: pipe.fileHandleForReading.fileDescriptor, upTo: outputCap) + return String(data: output, encoding: .utf8) } } diff --git a/TablePro/Core/Database/ProcessNativeDumpRunner.swift b/TablePro/Core/Database/ProcessNativeDumpRunner.swift index 3f212ddbfb..10cc9eb9f8 100644 --- a/TablePro/Core/Database/ProcessNativeDumpRunner.swift +++ b/TablePro/Core/Database/ProcessNativeDumpRunner.swift @@ -9,11 +9,9 @@ import Foundation final class ProcessNativeDumpRunner: NativeDumpRunner, @unchecked Sendable { private let command: NativeDumpCommand private let process = Process() - private let stderrPipe = Pipe() + private let stderrPipe: Pipe private let stderrReader: PipeReader private let stateLock = NSLock() - /// Separate from `stateLock`, which `cancel()` takes and which must never wait on a pipe. - private let stderrLock = NSLock() private var stderrBuffer = Data() private var wasCancelled = false private var terminationResult: NativeDumpRunResult? @@ -21,8 +19,9 @@ final class ProcessNativeDumpRunner: NativeDumpRunner, @unchecked Sendable { private var redirectedHandle: FileHandle? private var credentialsFileURL: URL? - init(command: NativeDumpCommand) { + init(command: NativeDumpCommand, stderrPipe: Pipe = Pipe()) { self.command = command + self.stderrPipe = stderrPipe stderrReader = PipeReader(stderrPipe.fileHandleForReading) } @@ -46,12 +45,9 @@ final class ProcessNativeDumpRunner: NativeDumpRunner, @unchecked Sendable { self.stderrReader.stop(drainingUpTo: stderrCap) self.releaseRedirection() - self.stderrLock.lock() + self.stateLock.lock() let stderrText = String(data: self.stderrBuffer, encoding: .utf8)? .trimmingCharacters(in: .whitespacesAndNewlines) ?? "" - self.stderrLock.unlock() - - self.stateLock.lock() let result = NativeDumpRunResult( exitCode: proc.terminationStatus, stderr: stderrText, @@ -78,8 +74,8 @@ final class ProcessNativeDumpRunner: NativeDumpRunner, @unchecked Sendable { } private func append(_ chunk: Data, cap: Int) { - stderrLock.lock() - defer { stderrLock.unlock() } + stateLock.lock() + defer { stateLock.unlock() } stderrBuffer.append(chunk) if stderrBuffer.count > cap { stderrBuffer = Data(stderrBuffer.suffix(cap)) diff --git a/TablePro/Core/Process/DescriptorRead.swift b/TablePro/Core/Process/DescriptorRead.swift index b05cf41fba..d604edb70c 100644 --- a/TablePro/Core/Process/DescriptorRead.swift +++ b/TablePro/Core/Process/DescriptorRead.swift @@ -6,18 +6,9 @@ import Darwin import Foundation -/// Reads a descriptor with `read(2)` and reports a failed read as a thrown `POSIXError`. -/// -/// `FileHandle.availableData` raises `NSFileHandleOperationException` on any failed read, and an -/// Objective-C exception unwinding through Swift ends the process. `FileHandle.read(upToCount:)` -/// throws instead, but on a pipe it waits for the whole count or for every writer to close, and on -/// a non-blocking pipe it throws away the bytes it read before `EAGAIN`, so it cannot return -/// whatever has arrived so far. -enum DescriptorRead { - /// The most a macOS pipe holds, so a writer that has exited cannot have left more than this. +internal enum DescriptorRead { static let pipeCapacity = 65_536 - /// The bytes one read returns, at most `limit`, empty at end of file. static func availableBytes(from descriptor: Int32, upTo limit: Int = pipeCapacity) throws -> Data { var bytes = Data(count: limit) let received = try bytes.withUnsafeMutableBytes { buffer in @@ -27,25 +18,52 @@ enum DescriptorRead { return bytes } - /// Whether a read would return at once, with bytes or at end of file, instead of waiting on a - /// writer. Asks `poll(2)` rather than setting `O_NONBLOCK`, which would change the descriptor for - /// every other reader of it too. + static func nextBytes(from descriptor: Int32, upTo limit: Int = pipeCapacity) throws -> Data { + while true { + do { + return try availableBytes(from: descriptor, upTo: limit) + } catch POSIXError.EAGAIN { + try waitForInput(descriptor) + } + } + } + + static func bufferedBytes(from descriptor: Int32, upTo limit: Int) -> Data { + var buffered = Data() + while buffered.count < limit, hasInputWithoutWaiting(descriptor) { + let wanted = min(pipeCapacity, limit - buffered.count) + guard let chunk = try? availableBytes(from: descriptor, upTo: wanted), !chunk.isEmpty else { break } + buffered.append(chunk) + } + return buffered + } + static func hasInputWithoutWaiting(_ descriptor: Int32) -> Bool { var request = pollfd(fd: descriptor, events: Int16(POLLIN), revents: 0) guard poll(&request, 1, 0) > 0 else { return false } return request.revents & Int16(POLLIN | POLLHUP) != 0 } + private static func waitForInput(_ descriptor: Int32) throws { + var request = pollfd(fd: descriptor, events: Int16(POLLIN), revents: 0) + while poll(&request, 1, -1) == -1 { + try throwUnlessInterrupted(errno) + } + } + private static func read(_ descriptor: Int32, into buffer: UnsafeMutableRawBufferPointer) throws -> Int { while true { let received = Darwin.read(descriptor, buffer.baseAddress, buffer.count) if received >= 0 { return received } - let failure = errno - guard failure == EINTR else { - throw POSIXError(POSIXErrorCode(rawValue: failure) ?? .EIO) - } + try throwUnlessInterrupted(errno) + } + } + + private static func throwUnlessInterrupted(_ failure: Int32) throws { + guard failure == EINTR else { + throw POSIXError(POSIXErrorCode(rawValue: failure) ?? .EIO) } } } diff --git a/TablePro/Core/Process/PipeReader.swift b/TablePro/Core/Process/PipeReader.swift index 82ccc69549..df92dc9c73 100644 --- a/TablePro/Core/Process/PipeReader.swift +++ b/TablePro/Core/Process/PipeReader.swift @@ -6,14 +6,7 @@ import Foundation import os -/// Hands a pipe's output to a consumer as it arrives, one chunk at a time and in order. -/// -/// Clearing `readabilityHandler` does not recall a callback Foundation has already dispatched, so -/// that callback can run its read after the owner has drained the pipe or closed it. Every read -/// here happens under one lock and only while a consumer is installed, and both ways of stopping -/// remove the consumer under that lock, so once either returns nothing is reading the descriptor -/// and nothing will. The handle can be closed straight after. -final class PipeReader: @unchecked Sendable { +internal final class PipeReader: @unchecked Sendable { private static let logger = Logger(subsystem: "com.TablePro", category: "PipeReader") private let handle: FileHandle @@ -33,24 +26,14 @@ final class PipeReader: @unchecked Sendable { } } - /// Stops reading, first handing the consumer what the pipe already holds, up to `limit` bytes. - /// It never waits on a writer, because a helper that inherited the write end can hold it open - /// for as long as it lives. func stop(drainingUpTo limit: Int = 0) { stopReading { consumer in - var taken = 0 - while taken < limit, DescriptorRead.hasInputWithoutWaiting(descriptor) { - let wanted = min(DescriptorRead.pipeCapacity, limit - taken) - guard let chunk = try? DescriptorRead.availableBytes(from: descriptor, upTo: wanted), - !chunk.isEmpty else { return } - taken += chunk.count - consumer(chunk) - } + let buffered = DescriptorRead.bufferedBytes(from: descriptor, upTo: limit) + guard !buffered.isEmpty else { return } + consumer(buffered) } } - /// Stops reading once every writer has closed the pipe, handing the consumer everything written - /// before that. A helper that inherited the write end keeps this waiting for as long as it lives. func stopAtEndOfFile() { stopReading { consumer in while let chunk = try? DescriptorRead.availableBytes(from: descriptor), !chunk.isEmpty { diff --git a/TablePro/Core/Process/SupervisedProcessRunner.swift b/TablePro/Core/Process/SupervisedProcessRunner.swift index aa27414c26..a436314113 100644 --- a/TablePro/Core/Process/SupervisedProcessRunner.swift +++ b/TablePro/Core/Process/SupervisedProcessRunner.swift @@ -144,11 +144,6 @@ final class ProcessSupervisedRunner: SupervisedProcessRunner, @unchecked Sendabl } } - /// The termination handler can run before the pipe delivers its last readability callback, so - /// what is still buffered is drained on the way out: a dropped final line is how a process that - /// announced itself ready right before exiting reads as one that never did. A callback that was - /// already dispatched has finished its delivery by the time the reader stops, so no line can - /// reach the stream after it has been closed. private func finish(exitCode: Int32) { stdoutReader.stop() stderrReader.stopAtEndOfFile() diff --git a/TableProTests/Core/MCP/Transport/BridgeStdinTests.swift b/TableProTests/Core/MCP/Transport/BridgeStdinTests.swift new file mode 100644 index 0000000000..ca0e703984 --- /dev/null +++ b/TableProTests/Core/MCP/Transport/BridgeStdinTests.swift @@ -0,0 +1,50 @@ +// +// BridgeStdinTests.swift +// TableProTests +// + +import Darwin +import Foundation +import Testing + +@testable import TablePro + +struct BridgeStdinTests { + @Test("A non-blocking stdin keeps the session reading until the host closes it", .timeLimit(.minutes(1))) + func nonBlockingStdinReadsUntilEndOfFile() async { + let pipe = Pipe() + defer { withExtendedLifetime(pipe) {} } + let descriptor = pipe.fileHandleForReading.fileDescriptor + #expect(fcntl(descriptor, F_SETFL, fcntl(descriptor, F_GETFL) | O_NONBLOCK) != -1) + BackgroundPipeWriter.write( + [Data("{\"id\":1}\n".utf8), Data("{\"id\":2}\n".utf8)], + to: pipe.fileHandleForWriting, + pausingBeforeEach: 0.2 + ) + let logger = RecordingBridgeLogger() + + var lines: [Data] = [] + for await line in BridgeStdin.lines(from: pipe.fileHandleForReading, logger: logger) { + lines.append(line) + } + + #expect(lines == [Data("{\"id\":1}".utf8), Data("{\"id\":2}".utf8)]) + #expect(logger.entries.isEmpty) + } + + @Test("A stdin that cannot be read ends the session and says why", .timeLimit(.minutes(1))) + func unreadableStdinEndsTheSession() async throws { + let descriptor = open(FileManager.default.temporaryDirectory.path, O_RDONLY) + try #require(descriptor >= 0) + let directory = FileHandle(fileDescriptor: descriptor, closeOnDealloc: true) + let logger = RecordingBridgeLogger() + + var lines: [Data] = [] + for await line in BridgeStdin.lines(from: directory, logger: logger) { + lines.append(line) + } + + #expect(lines.isEmpty) + #expect(logger.entries.map(\.level) == [.error]) + } +} diff --git a/TableProTests/Core/Process/DescriptorReadTests.swift b/TableProTests/Core/Process/DescriptorReadTests.swift index 2999df0962..2d2b78f447 100644 --- a/TableProTests/Core/Process/DescriptorReadTests.swift +++ b/TableProTests/Core/Process/DescriptorReadTests.swift @@ -10,13 +10,16 @@ import Testing @testable import TablePro struct DescriptorReadTests { - @Test("A read returns what has arrived without waiting for the rest") - func returnsWhatHasArrived() throws { + @Test("A read returns what has arrived without waiting for the rest", .timeLimit(.minutes(1))) + func returnsWhatHasArrived() async throws { let pipe = Pipe() defer { withExtendedLifetime(pipe) {} } try pipe.fileHandleForWriting.write(contentsOf: Data("first".utf8)) + let descriptor = pipe.fileHandleForReading.fileDescriptor - let bytes = try DescriptorRead.availableBytes(from: pipe.fileHandleForReading.fileDescriptor) + let bytes = await HeldOpenWriter(pipe.fileHandleForWriting).finishedOnItsOwnThread { + try? DescriptorRead.availableBytes(from: descriptor) + } ?? nil #expect(bytes == Data("first".utf8)) } @@ -66,11 +69,6 @@ struct DescriptorReadTests { #expect(error?.code == .EBADF) } - /// End of file comes once every holder of the write end has let go, and a child this process is - /// spawning at that moment briefly holds it too. Measured: with four threads launching - /// `/usr/bin/true`, 14 of 20,000 closes were not yet end of file at once, and 0 of 20,000 - /// without them; beside the other process suites here, 9 of 200 were not, and all 200 were - /// 200ms later. So the last check waits for it rather than expecting it on the spot. @Test("Only a pipe holding bytes or at end of file reads without waiting", .timeLimit(.minutes(1))) func readinessFollowsThePipe() async throws { let pipe = Pipe() @@ -91,4 +89,41 @@ struct DescriptorReadTests { } #expect(DescriptorRead.hasInputWithoutWaiting(descriptor)) } + + @Test("On a non-blocking pipe the next read waits for bytes instead of failing with EAGAIN", .timeLimit(.minutes(1))) + func nextBytesWaitsOnANonBlockingPipe() throws { + let pipe = Pipe() + defer { withExtendedLifetime(pipe) {} } + let descriptor = pipe.fileHandleForReading.fileDescriptor + #expect(fcntl(descriptor, F_SETFL, fcntl(descriptor, F_GETFL) | O_NONBLOCK) != -1) + BackgroundPipeWriter.write([Data("late".utf8)], to: pipe.fileHandleForWriting, pausingBeforeEach: 0.2) + + #expect(try DescriptorRead.nextBytes(from: descriptor) == Data("late".utf8)) + #expect(try DescriptorRead.nextBytes(from: descriptor).isEmpty) + } + + @Test("Buffered bytes stop at the limit and never wait on a writer that is still open", .timeLimit(.minutes(1))) + func bufferedBytesTakeWhatThePipeHolds() async throws { + let pipe = Pipe() + defer { withExtendedLifetime(pipe) {} } + let descriptor = pipe.fileHandleForReading.fileDescriptor + let written = Data((0 ..< 10_000).map { UInt8($0 % 251) }) + try pipe.fileHandleForWriting.write(contentsOf: written) + let writer = HeldOpenWriter(pipe.fileHandleForWriting) + + let first = await writer.finishedOnItsOwnThread { + DescriptorRead.bufferedBytes(from: descriptor, upTo: 4_096) + } + let rest = await writer.finishedOnItsOwnThread { + DescriptorRead.bufferedBytes(from: descriptor, upTo: DescriptorRead.pipeCapacity) + } + let nothing = await writer.finishedOnItsOwnThread { + DescriptorRead.bufferedBytes(from: descriptor, upTo: DescriptorRead.pipeCapacity) + } + + #expect(first == written.prefix(4_096)) + #expect(rest == written.dropFirst(4_096)) + #expect(nothing == Data()) + #expect(fcntl(descriptor, F_GETFL) & O_NONBLOCK == 0) + } } diff --git a/TableProTests/Core/Process/PipeReaderTests.swift b/TableProTests/Core/Process/PipeReaderTests.swift index 4b7173b47d..ed838ecb91 100644 --- a/TableProTests/Core/Process/PipeReaderTests.swift +++ b/TableProTests/Core/Process/PipeReaderTests.swift @@ -56,21 +56,14 @@ struct PipeReaderTests { reader.start { received.append($0) } let payload = Data((0 ..< 3 * DescriptorRead.pipeCapacity).map { UInt8($0 % 251) }) - let writer = pipe.fileHandleForWriting - Thread.detachNewThread { - try? writer.write(contentsOf: payload) - try? writer.close() - } + BackgroundPipeWriter.write([payload], to: pipe.fileHandleForWriting, pausingBeforeEach: 0) try await waitUntil { received.data.count == payload.count && handle.readabilityHandler == nil } #expect(received.data == payload) } - /// The order that crashed a dump: the process exits, the termination handler drains the pipe - /// while a readability callback Foundation had already dispatched waits, and the callback then - /// reads a pipe that is empty with its writer still open. @Test("A callback dispatched before a drain reads nothing when it runs after it", .timeLimit(.minutes(1))) - func dispatchedCallbackAfterDrainReadsNothing() throws { + func dispatchedCallbackAfterDrainReadsNothing() async throws { let pipe = Pipe() defer { withExtendedLifetime(pipe) {} } let handle = pipe.fileHandleForReading @@ -80,16 +73,19 @@ struct PipeReaderTests { let dispatched = try #require(handle.readabilityHandler) try pipe.fileHandleForWriting.write(contentsOf: Data("early".utf8)) - reader.stop(drainingUpTo: DescriptorRead.pipeCapacity) + let drained: Void? = await HeldOpenWriter(pipe.fileHandleForWriting).finishedOnItsOwnThread { + reader.stop(drainingUpTo: DescriptorRead.pipeCapacity) + } + try #require(drained != nil) #expect(received.data == Data("early".utf8)) - dispatched(handle) try pipe.fileHandleForWriting.write(contentsOf: Data("late".utf8)) dispatched(handle) #expect(received.data == Data("early".utf8)) let descriptor = handle.fileDescriptor #expect(fcntl(descriptor, F_GETFL) & O_NONBLOCK == 0) + try pipe.fileHandleForWriting.close() #expect(try DescriptorRead.availableBytes(from: descriptor) == Data("late".utf8)) } @@ -114,8 +110,11 @@ struct PipeReaderTests { reader.stop() } - @Test("A drain takes what the pipe holds up to its limit and leaves the descriptor blocking") - func drainStopsAtTheLimitWithoutWaitingOnTheWriter() throws { + @Test( + "A drain takes what the pipe holds up to its limit and leaves the descriptor blocking", + .timeLimit(.minutes(1)) + ) + func drainStopsAtTheLimitWithoutWaitingOnTheWriter() async throws { let pipe = Pipe() defer { withExtendedLifetime(pipe) {} } let handle = pipe.fileHandleForReading @@ -126,11 +125,15 @@ struct PipeReaderTests { let buffered = Data((0 ..< 10_000).map { UInt8($0 % 251) }) try pipe.fileHandleForWriting.write(contentsOf: buffered) - reader.stop(drainingUpTo: 4_096) + let drained: Void? = await HeldOpenWriter(pipe.fileHandleForWriting).finishedOnItsOwnThread { + reader.stop(drainingUpTo: 4_096) + } + #expect(drained != nil) #expect(received.data == buffered.prefix(4_096)) let descriptor = handle.fileDescriptor #expect(fcntl(descriptor, F_GETFL) & O_NONBLOCK == 0) + try pipe.fileHandleForWriting.close() #expect(try DescriptorRead.availableBytes(from: descriptor) == buffered.dropFirst(4_096)) } @@ -144,13 +147,8 @@ struct PipeReaderTests { reader.start { received.append($0) } handle.readabilityHandler = nil - let writer = pipe.fileHandleForWriting - try writer.write(contentsOf: Data("before".utf8)) - Thread.detachNewThread { - Thread.sleep(forTimeInterval: 0.2) - try? writer.write(contentsOf: Data(" after".utf8)) - try? writer.close() - } + try pipe.fileHandleForWriting.write(contentsOf: Data("before".utf8)) + BackgroundPipeWriter.write([Data(" after".utf8)], to: pipe.fileHandleForWriting, pausingBeforeEach: 0.2) reader.stopAtEndOfFile() #expect(received.data == Data("before after".utf8)) diff --git a/TableProTests/Core/Services/Infrastructure/PreConnectHookRunnerTests.swift b/TableProTests/Core/Services/Infrastructure/PreConnectHookRunnerTests.swift index 216544241e..4bee6e47af 100644 --- a/TableProTests/Core/Services/Infrastructure/PreConnectHookRunnerTests.swift +++ b/TableProTests/Core/Services/Infrastructure/PreConnectHookRunnerTests.swift @@ -22,10 +22,6 @@ struct PreConnectHookRunnerTests { } } - /// A script that writes its reason and exits at once can still have it in the pipe when the - /// runner reads what it collected. Measured with sixteen scripts at a time: 352 of 16,000 - /// failures came back without their message before the pipe was drained on the way out, and 0 - /// of 16,000 after. @Test("A failing script's message reaches the error every time", .timeLimit(.minutes(1))) func failureCarriesTheScriptsMessage() async { for _ in 0 ..< 4 { diff --git a/TableProTests/Database/ProcessNativeDumpRunnerTests.swift b/TableProTests/Database/ProcessNativeDumpRunnerTests.swift index 22e617c6ee..b384767597 100644 --- a/TableProTests/Database/ProcessNativeDumpRunnerTests.swift +++ b/TableProTests/Database/ProcessNativeDumpRunnerTests.swift @@ -3,6 +3,7 @@ // TableProTests // +import Darwin import Foundation import Testing @@ -67,4 +68,42 @@ struct ProcessNativeDumpRunnerTests { #expect(result.exitCode == 4) #expect(result.stderr.utf8.count <= cap) } + + @Test( + "A stderr callback dispatched before the tool exited leaves the pipe alone once the result is in", + .timeLimit(.minutes(1)) + ) + func lateStderrCallbackLeavesThePipeAlone() async throws { + let gate = FileManager.default.temporaryDirectory + .appendingPathComponent("tablepro-dump-gate-\(UUID().uuidString)") + defer { try? FileManager.default.removeItem(at: gate) } + let pipe = Pipe() + defer { withExtendedLifetime(pipe) {} } + let survivingDescriptor = dup(pipe.fileHandleForWriting.fileDescriptor) + try #require(survivingDescriptor >= 0) + let survivingWriter = FileHandle(fileDescriptor: survivingDescriptor, closeOnDealloc: true) + defer { try? survivingWriter.close() } + + let runner = ProcessNativeDumpRunner( + command: command("while [ ! -e '\(gate.path)' ]; do sleep 0.01; done; printf '%s' 'refused' >&2; exit 5"), + stderrPipe: pipe + ) + try runner.start() + let handle = pipe.fileHandleForReading + let dispatched = try #require(handle.readabilityHandler) + #expect(FileManager.default.createFile(atPath: gate.path, contents: nil)) + + let result = try #require(await HeldOpenWriter(survivingWriter).finished { await runner.result }) + #expect(result.exitCode == 5) + #expect(result.stderr == "refused") + + try survivingWriter.write(contentsOf: Data("late".utf8)) + dispatched(handle) + + let descriptor = handle.fileDescriptor + #expect(handle.readabilityHandler == nil) + #expect(fcntl(descriptor, F_GETFL) & O_NONBLOCK == 0) + try survivingWriter.close() + #expect(try DescriptorRead.availableBytes(from: descriptor) == Data("late".utf8)) + } } diff --git a/TableProTests/Helpers/BackgroundPipeWriter.swift b/TableProTests/Helpers/BackgroundPipeWriter.swift new file mode 100644 index 0000000000..82caccb9fa --- /dev/null +++ b/TableProTests/Helpers/BackgroundPipeWriter.swift @@ -0,0 +1,20 @@ +// +// BackgroundPipeWriter.swift +// TableProTests +// + +import Darwin +import Foundation + +internal enum BackgroundPipeWriter { + static func write(_ chunks: [Data], to writer: FileHandle, pausingBeforeEach pause: TimeInterval) { + _ = fcntl(writer.fileDescriptor, F_SETNOSIGPIPE, 1) + Thread.detachNewThread { + for chunk in chunks { + Thread.sleep(forTimeInterval: pause) + try? writer.write(contentsOf: chunk) + } + try? writer.close() + } + } +} diff --git a/TableProTests/Helpers/HeldOpenWriter.swift b/TableProTests/Helpers/HeldOpenWriter.swift new file mode 100644 index 0000000000..470a6eaeeb --- /dev/null +++ b/TableProTests/Helpers/HeldOpenWriter.swift @@ -0,0 +1,46 @@ +// +// HeldOpenWriter.swift +// TableProTests +// + +import Foundation + +internal struct HeldOpenWriter: Sendable { + private enum Arrival: Sendable { + case finished(Value) + case deadlinePassed + } + + private static let deadline = Duration.seconds(10) + + private let handle: FileHandle + + init(_ handle: FileHandle) { + self.handle = handle + } + + func finished(_ work: @escaping @Sendable () async -> Value) async -> Value? { + let handle = handle + return await withTaskGroup(of: Arrival.self) { group in + group.addTask { .finished(await work()) } + group.addTask { + try? await Task.sleep(for: Self.deadline) + return .deadlinePassed + } + guard case .finished(let value)? = await group.next() else { + try? handle.close() + return nil + } + group.cancelAll() + return value + } + } + + func finishedOnItsOwnThread(_ work: @escaping @Sendable () -> Value) async -> Value? { + await finished { + await withCheckedContinuation { continuation in + Thread.detachNewThread { continuation.resume(returning: work()) } + } + } + } +}