fix(mcp): wait for room on a non-blocking stdout or stderr in tablepro-mcp, and bound every pipe test that could hang

CHANGELOG.md

@@ -130,6 +130,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Pre-connect script failures sometimes reported without the script's own error message.
 - Failed MongoDB statements, including writes the server rejected, reported as successful with an empty result.
 - `tablepro-mcp` crashing when its standard input was non-blocking.
+- `tablepro-mcp` using a full CPU core, or crashing, when its standard output or error 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/BridgeMain.swift

@@ -82,6 +82,6 @@ struct TableProMcpBridge {
             )
         )
         guard let data = try? JsonRpcCodec.encodeLine(envelope) else { return }
-        FileHandle.standardOutput.write(data)
+        try? DescriptorWrite.allBytes(data, to: FileHandle.standardOutput.fileDescriptor)
     }
 }

TablePro/CLI/BridgeProxy.swift

@@ -142,7 +142,7 @@ actor BridgeProxy {
         self.upstream = upstream
         self.discovery = discovery
         self.logger = logger
-        self.stdout = BridgeStdout(handle: stdout)
+        self.stdout = BridgeStdout(handle: stdout, logger: logger)
         self.hostLines = BridgeStdin.lines(from: stdin, logger: logger)
     }
 
@@ -564,18 +564,20 @@ enum BridgeStdin {
 
 actor BridgeStdout {
     private let handle: FileHandle
+    private let logger: any MCPBridgeLogger
 
-    init(handle: FileHandle) {
+    init(handle: FileHandle, logger: any MCPBridgeLogger) {
         self.handle = handle
+        self.logger = logger
     }
 
     func write(_ payload: Data) {
         var line = payload
         line.append(0x0A)
         do {
-            try handle.write(contentsOf: line)
+            try DescriptorWrite.allBytes(line, to: handle.fileDescriptor)
         } catch {
-            FileHandle.standardError.write(Data("[error] stdout write failed: \(error)\n".utf8))
+            logger.log(.error, "Writing stdout failed: \(error.localizedDescription)")
         }
     }
 }

TablePro/Core/Database/CLIToolVersionProbe.swift

@@ -6,22 +6,13 @@
 import Foundation
 import os
 
-/// Asks a command line tool what it is, by running it with `--version`.
-///
-/// Synchronous on purpose: `NativeDumpDescriptor.CommandLineTool`'s resolution hooks are
-/// synchronous closures that `NativeDumpService` already runs inside a detached task, so an async
-/// probe would have to change every one of them.
 enum CLIToolVersionProbe {
     private static let logger = Logger(subsystem: "com.TablePro", category: "CLIToolVersionProbe")
 
     static let defaultTimeout: TimeInterval = 3
 
-    /// A version banner is one line. Reading past this is a tool doing something other than
-    /// answering the question, and the answer is taken from what arrived rather than waited for.
     static let outputCap = 64 * 1_024
 
-    /// Standard output of `<path> --version`, or nil when the tool cannot run, does not answer in
-    /// time, or exits non-zero.
     static func versionOutput(of path: String, timeout: TimeInterval = defaultTimeout) -> String? {
         let process = Process()
         process.executableURL = URL(fileURLWithPath: path)

TablePro/Core/MCP/Transport/MCPBridgeLogger.swift

@@ -36,7 +36,15 @@ public struct MCPOSBridgeLogger: MCPBridgeLogger {
 public struct MCPStderrBridgeLogger: MCPBridgeLogger {
     private static let lock = NSLock()
 
-    public init() {}
+    private let descriptor: Int32
+
+    public init() {
+        self.init(descriptor: FileHandle.standardError.fileDescriptor)
+    }
+
+    internal init(descriptor: Int32) {
+        self.descriptor = descriptor
+    }
 
     public func log(_ level: MCPBridgeLogLevel, _ message: String) {
         let prefix: String
@@ -50,7 +58,7 @@ public struct MCPStderrBridgeLogger: MCPBridgeLogger {
         guard let data = payload.data(using: .utf8) else { return }
         Self.lock.lock()
         defer { Self.lock.unlock() }
-        FileHandle.standardError.write(data)
+        try? DescriptorWrite.allBytes(data, to: descriptor)
     }
 }
 

TablePro/Core/Process/DescriptorWrite.swift

@@ -0,0 +1,42 @@
+//
+//  DescriptorWrite.swift
+//  TablePro
+//
+
+import Darwin
+import Foundation
+
+internal enum DescriptorWrite {
+    static func allBytes(_ bytes: Data, to descriptor: Int32) throws {
+        try bytes.withUnsafeBytes { buffer in
+            guard let start = buffer.baseAddress else { return }
+            var offset = 0
+            while offset < buffer.count {
+                let written = Darwin.write(descriptor, start + offset, buffer.count - offset)
+                if written >= 0 {
+                    offset += written
+                    continue
+                }
+                let failure = errno
+                if failure == EAGAIN {
+                    try waitForRoom(descriptor)
+                    continue
+                }
+                try throwUnlessInterrupted(failure)
+            }
+        }
+    }
+
+    private static func waitForRoom(_ descriptor: Int32) throws {
+        var request = pollfd(fd: descriptor, events: Int16(POLLOUT), revents: 0)
+        while poll(&request, 1, -1) == -1 {
+            try throwUnlessInterrupted(errno)
+        }
+    }
+
+    private static func throwUnlessInterrupted(_ failure: Int32) throws {
+        guard failure == EINTR else {
+            throw POSIXError(POSIXErrorCode(rawValue: failure) ?? .EIO)
+        }
+    }
+}

TableProTests/Core/MCP/Transport/BridgeStdinTests.swift

@@ -10,6 +10,16 @@ import Testing
 @testable import TablePro
 
 struct BridgeStdinTests {
+    private static func everyLine(of stream: AsyncStream<Data>) async -> [Data]? {
+        await BoundedCall.result {
+            var lines: [Data] = []
+            for await line in stream {
+                lines.append(line)
+            }
+            return lines
+        }
+    }
+
     @Test("A non-blocking stdin keeps the session reading until the host closes it", .timeLimit(.minutes(1)))
     func nonBlockingStdinReadsUntilEndOfFile() async {
         let pipe = Pipe()
@@ -23,10 +33,7 @@ struct BridgeStdinTests {
         )
         let logger = RecordingBridgeLogger()
 
-        var lines: [Data] = []
-        for await line in BridgeStdin.lines(from: pipe.fileHandleForReading, logger: logger) {
-            lines.append(line)
-        }
+        let lines = await Self.everyLine(of: BridgeStdin.lines(from: pipe.fileHandleForReading, logger: logger))
 
         #expect(lines == [Data("{\"id\":1}".utf8), Data("{\"id\":2}".utf8)])
         #expect(logger.entries.isEmpty)
@@ -39,12 +46,9 @@ struct BridgeStdinTests {
         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)
-        }
+        let lines = await Self.everyLine(of: BridgeStdin.lines(from: directory, logger: logger))
 
-        #expect(lines.isEmpty)
+        #expect(lines?.isEmpty == true)
         #expect(logger.entries.map(\.level) == [.error])
     }
 }

TableProTests/Core/MCP/Transport/BridgeStdoutTests.swift

@@ -0,0 +1,48 @@
+//
+//  BridgeStdoutTests.swift
+//  TableProTests
+//
+
+import Darwin
+import Foundation
+import Testing
+
+@testable import TablePro
+
+struct BridgeStdoutTests {
+    @Test("A line larger than the pipe reaches a non-blocking stdout whole", .timeLimit(.minutes(1)))
+    func nonBlockingStdoutGetsTheWholeLine() async throws {
+        let pipe = Pipe()
+        defer { withExtendedLifetime(pipe) {} }
+        let descriptor = pipe.fileHandleForWriting.fileDescriptor
+        #expect(fcntl(descriptor, F_SETFL, fcntl(descriptor, F_GETFL) | O_NONBLOCK) != -1)
+        #expect(fcntl(descriptor, F_SETNOSIGPIPE, 1) != -1)
+        let payload = Data(repeating: UInt8(ascii: "a"), count: 3 * DescriptorRead.pipeCapacity)
+        let logger = RecordingBridgeLogger()
+        let stdout = BridgeStdout(handle: pipe.fileHandleForWriting, logger: logger)
+
+        async let drained = BackgroundPipeReader.everything(from: pipe.fileHandleForReading, pausingFirst: 0.2)
+        let wrote: Void? = await BoundedCall.result { await stdout.write(payload) }
+        try pipe.fileHandleForWriting.close()
+
+        let received = try #require(await drained)
+        #expect(wrote != nil)
+        #expect(received == payload + Data([0x0A]))
+        #expect(logger.entries.isEmpty)
+    }
+
+    @Test("A stdout nobody reads says why through the bridge's logger", .timeLimit(.minutes(1)))
+    func unwritableStdoutIsLogged() async throws {
+        let pipe = Pipe()
+        defer { withExtendedLifetime(pipe) {} }
+        #expect(fcntl(pipe.fileHandleForWriting.fileDescriptor, F_SETNOSIGPIPE, 1) != -1)
+        try pipe.fileHandleForReading.close()
+        let logger = RecordingBridgeLogger()
+        let stdout = BridgeStdout(handle: pipe.fileHandleForWriting, logger: logger)
+
+        let wrote: Void? = await BoundedCall.result { await stdout.write(Data("{}".utf8)) }
+
+        #expect(wrote != nil)
+        #expect(logger.entries.map(\.level) == [.error])
+    }
+}

TableProTests/Core/MCP/Transport/MCPStderrBridgeLoggerTests.swift

@@ -0,0 +1,46 @@
+//
+//  MCPStderrBridgeLoggerTests.swift
+//  TableProTests
+//
+
+import Darwin
+import Foundation
+import Testing
+
+@testable import TablePro
+
+struct MCPStderrBridgeLoggerTests {
+    private static func fill(_ descriptor: Int32) -> Int {
+        let chunk = [UInt8](repeating: UInt8(ascii: "x"), count: DescriptorRead.pipeCapacity)
+        var total = 0
+        while true {
+            let written = chunk.withUnsafeBytes { Darwin.write(descriptor, $0.baseAddress, $0.count) }
+            guard written > 0 else { return total }
+            total += written
+        }
+    }
+
+    @Test("A log line waits for room on a full non-blocking stderr", .timeLimit(.minutes(1)))
+    func fullNonBlockingStderrGetsTheLine() async throws {
+        let pipe = Pipe()
+        defer { withExtendedLifetime(pipe) {} }
+        let descriptor = pipe.fileHandleForWriting.fileDescriptor
+        #expect(fcntl(descriptor, F_SETFL, fcntl(descriptor, F_GETFL) | O_NONBLOCK) != -1)
+        #expect(fcntl(descriptor, F_SETNOSIGPIPE, 1) != -1)
+        let filled = Self.fill(descriptor)
+        #expect(filled > 0)
+        let logger = MCPStderrBridgeLogger(descriptor: descriptor)
+
+        async let drained = BackgroundPipeReader.everything(from: pipe.fileHandleForReading, pausingFirst: 0.2)
+        let logged: Void? = await BoundedCall.resultOnItsOwnThread {
+            logger.log(.error, "Upstream stream ended")
+        }
+        try pipe.fileHandleForWriting.close()
+
+        let received = try #require(await drained)
+        let line = Data("[error] Upstream stream ended\n".utf8)
+        #expect(logged != nil)
+        #expect(received.count == filled + line.count)
+        #expect(received.suffix(line.count) == line)
+    }
+}

TableProTests/Core/Process/DescriptorReadTests.swift

@@ -29,6 +29,7 @@ struct DescriptorReadTests {
         let pipe = Pipe()
         defer { withExtendedLifetime(pipe) {} }
         try pipe.fileHandleForWriting.write(contentsOf: Data("abcdef".utf8))
+        try pipe.fileHandleForWriting.close()
         let descriptor = pipe.fileHandleForReading.fileDescriptor
 
         #expect(try DescriptorRead.availableBytes(from: descriptor, upTo: 4) == Data("abcd".utf8))
@@ -46,15 +47,22 @@ struct DescriptorReadTests {
         #expect(bytes.isEmpty)
     }
 
-    @Test("An empty non-blocking pipe is a thrown EAGAIN, the read that raised in a dump's stderr callback")
-    func emptyNonBlockingPipeThrows() throws {
+    @Test(
+        "An empty non-blocking pipe is a thrown EAGAIN, the read that raised in a dump's stderr callback",
+        .timeLimit(.minutes(1))
+    )
+    func emptyNonBlockingPipeThrows() async throws {
         let pipe = Pipe()
         defer { withExtendedLifetime(pipe) {} }
         let descriptor = pipe.fileHandleForReading.fileDescriptor
         #expect(fcntl(descriptor, F_SETFL, fcntl(descriptor, F_GETFL) | O_NONBLOCK) != -1)
 
+        let outcome = await HeldOpenWriter(pipe.fileHandleForWriting).finishedOnItsOwnThread {
+            Result { try DescriptorRead.availableBytes(from: descriptor) }
+        }
+        let read = try #require(outcome)
         let error = #expect(throws: POSIXError.self) {
-            try DescriptorRead.availableBytes(from: descriptor)
+            try read.get()
         }
 
         #expect(error?.code == .EAGAIN)
@@ -79,7 +87,7 @@ struct DescriptorReadTests {
         try pipe.fileHandleForWriting.write(contentsOf: Data("x".utf8))
         #expect(DescriptorRead.hasInputWithoutWaiting(descriptor))
 
-        _ = try DescriptorRead.availableBytes(from: descriptor)
+        #expect(try DescriptorRead.availableBytes(from: descriptor, upTo: 1) == Data("x".utf8))
         #expect(!DescriptorRead.hasInputWithoutWaiting(descriptor))
 
         try pipe.fileHandleForWriting.close()
@@ -91,15 +99,22 @@ struct DescriptorReadTests {
     }
 
     @Test("On a non-blocking pipe the next read waits for bytes instead of failing with EAGAIN", .timeLimit(.minutes(1)))
-    func nextBytesWaitsOnANonBlockingPipe() throws {
+    func nextBytesWaitsOnANonBlockingPipe() async 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)
+        let arrived = try #require(await BoundedCall.resultOnItsOwnThread {
+            Result { try DescriptorRead.nextBytes(from: descriptor) }
+        })
+        #expect(try arrived.get() == Data("late".utf8))
+
+        let ended = try #require(await BoundedCall.resultOnItsOwnThread {
+            Result { try DescriptorRead.nextBytes(from: descriptor) }
+        })
+        #expect(try ended.get().isEmpty)
     }
 
     @Test("Buffered bytes stop at the limit and never wait on a writer that is still open", .timeLimit(.minutes(1)))