fix(mcp): keep tablepro-mcp reading a non-blocking stdin and drain every pipe without O_NONBLOCK

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.

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

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)
     }
 }

TablePro/Core/Database/ProcessNativeDumpRunner.swift

@@ -9,20 +9,19 @@ 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?
     private var continuation: CheckedContinuation<NativeDumpRunResult, Never>?
     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))

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)
         }
     }
 }

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 {

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()

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])
+    }
+}

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)
+    }
 }