From 64ad7593f055f4a5b91c5bdfc137cfd9c4242c28 Mon Sep 17 00:00:00 2001 From: Kazuki Nakashima <65545348+lynnswap@users.noreply.github.com> Date: Wed, 15 Jul 2026 18:48:40 +0900 Subject: [PATCH 1/2] fix(stdio): move input reads off cooperative executor FileHandle.AsyncBytes can block the sole cooperative worker in low-capacity CI runners, starving the task that writes and closes adapter stdin. Route input through a bounded DispatchIO channel while preserving EOF drain, read-error abort, and deterministic shutdown contracts. --- .../XcodeMCPProxyKit/Stdio/StdioAdapter.swift | 71 ++---- .../Stdio/StdioInputChannel.swift | 212 ++++++++++++++++++ .../StdioAdapterIntegrationTests.swift | 126 ++++++++++- .../StdioInputChannelTests.swift | 86 +++++++ 4 files changed, 441 insertions(+), 54 deletions(-) create mode 100644 Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift create mode 100644 Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift diff --git a/Sources/XcodeMCPProxyKit/Stdio/StdioAdapter.swift b/Sources/XcodeMCPProxyKit/Stdio/StdioAdapter.swift index f1ebef5..b7d87ae 100644 --- a/Sources/XcodeMCPProxyKit/Stdio/StdioAdapter.swift +++ b/Sources/XcodeMCPProxyKit/Stdio/StdioAdapter.swift @@ -1,40 +1,8 @@ import Foundation import Logging -import Synchronization import XcodeMCPKit import XcodeMCPProxyRuntime -private final class StdioReadIteratorReadiness: Sendable { - private struct State { - var isReady = false - var waiters: [CheckedContinuation] = [] - } - - private let state = Mutex(State()) - - func markReady() { - let waiters: [CheckedContinuation] = state.withLock { state in - guard state.isReady == false else { return [] } - state.isReady = true - let waiters = state.waiters - state.waiters.removeAll() - return waiters - } - for waiter in waiters { waiter.resume() } - } - - func waitUntilReady() async { - await withCheckedContinuation { continuation in - let shouldResume = state.withLock { state in - guard state.isReady == false else { return true } - state.waiters.append(continuation) - return false - } - if shouldResume { continuation.resume() } - } - } -} - package struct StdioAdapterShutdownPolicy: Sendable { package let requestDrainTimeout: Duration package let requestDrainPollInterval: Duration @@ -65,12 +33,11 @@ actor StdioAdapter { } private let requestTimeout: Duration? - private let inputHandle: FileHandle + private let inputReader: any StdioInputReading private let outputWriter: StdioWriter private let logger: Logger private let authority: MCPClientSessionAuthority private let shutdownPolicy: StdioAdapterShutdownPolicy - private let readIteratorReadiness = StdioReadIteratorReadiness() private var framer = StdioFramer() private var requestTasks: [UUID: Task] = [:] @@ -130,9 +97,25 @@ actor StdioAdapter { output: FileHandle, recipe: MCPTransportRecipe, shutdownPolicy: StdioAdapterShutdownPolicy + ) { + self.init( + requestTimeout: requestTimeout, + inputReader: StdioInputChannel(handle: input), + output: output, + recipe: recipe, + shutdownPolicy: shutdownPolicy + ) + } + + init( + requestTimeout: Duration?, + inputReader: any StdioInputReading, + output: FileHandle, + recipe: MCPTransportRecipe, + shutdownPolicy: StdioAdapterShutdownPolicy ) { self.requestTimeout = requestTimeout - self.inputHandle = input + self.inputReader = inputReader let logger = ProxyLogging.make("stdio.adapter") self.logger = logger self.outputWriter = StdioWriter(handle: output, logger: logger) @@ -152,18 +135,13 @@ actor StdioAdapter { } lifecycle = .running startAuthorityEventTask() - let input = inputHandle - let readIteratorReadiness = readIteratorReadiness + let inputReader = inputReader readTask = Task { [weak self] in - var iterator = input.bytes.makeAsyncIterator() - // Do not signal readiness before iterator creation: Foundation accesses the - // descriptor here and raises NSFileHandleOperationException if close wins. - readIteratorReadiness.markReady() var terminalError: (any Error)? do { - while let byte = try await iterator.next() { + while let chunk = try await inputReader.read() { guard Task.isCancelled == false else { break } - await self?.handleInput(Data([byte])) + await self?.handleInput(chunk) } } catch is CancellationError { // Explicit stop owns completion. @@ -198,6 +176,7 @@ actor StdioAdapter { isolated deinit { readTask?.cancel() + inputReader.stop() authorityEventTask?.cancel() for task in requestTasks.values { task.cancel() } closeTask?.cancel() @@ -385,10 +364,8 @@ private extension StdioAdapter { } func performClose(drainsOutput: Bool) async { - if readTask != nil { - await readIteratorReadiness.waitUntilReady() - try? inputHandle.close() - } + inputReader.stop() + await inputReader.waitUntilStopped() await authority.close() await drainRequestTasks() let eventTask = authorityEventTask diff --git a/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift b/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift new file mode 100644 index 0000000..f8b3e65 --- /dev/null +++ b/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift @@ -0,0 +1,212 @@ +import Darwin +import Dispatch +import Foundation +import XcodeMCPKit + +protocol StdioInputReading: Sendable { + func read() async throws -> Data? + func stop() + func waitUntilStopped() async +} + +private struct StdioInputReadError: Error, CustomStringConvertible, Sendable { + let code: Int32 + + var description: String { + "STDIO input read failed with errno \(code)" + } +} + +final class StdioInputChannel: StdioInputReading, @unchecked Sendable { + private enum Lifecycle { + case open + case endOfFile + case failed(StdioInputReadError) + case stopped + } + + private static let maxReadOperationByteCount = 64 * 1024 + + private let callbackQueue = DispatchQueue(label: "XcodeMCPProxy.StdioInputChannel.io") + private let channel: DispatchIO + private let terminal: AsyncTerminalSignal + + // Access is confined to callbackQueue. + private var lifecycle: Lifecycle = .open + private var bufferedInput = Data() + private var pendingRead: CheckedContinuation? + private var isReadOperationActive = false + private var activeOperationByteCount = 0 + + init(handle: FileHandle) { + let descriptor = dup(handle.fileDescriptor) + precondition(descriptor >= 0, "STDIO input FileHandle must be open") + + let terminal = AsyncTerminalSignal() + self.terminal = terminal + self.channel = DispatchIO( + type: .stream, + fileDescriptor: descriptor, + queue: callbackQueue + ) { _ in + _ = Darwin.close(descriptor) + try? handle.close() + terminal.signal() + } + channel.setLimit(lowWater: 1) + channel.setLimit(highWater: Self.maxReadOperationByteCount) + } + + func read() async throws -> Data? { + try await withTaskCancellationHandler { + try Task.checkCancellation() + return try await withCheckedThrowingContinuation { continuation in + callbackQueue.async { [self] in + registerRead(continuation) + } + } + } onCancel: { + stop() + } + } + + func stop() { + callbackQueue.async { [self] in + guard case .open = lifecycle else { return } + lifecycle = .stopped + bufferedInput.removeAll() + let continuation = pendingRead + pendingRead = nil + channel.close(flags: .stop) + continuation?.resume(throwing: CancellationError()) + } + } + + func waitUntilStopped() async { + await terminal.wait() + } + + private func registerRead( + _ continuation: CheckedContinuation + ) { + precondition(pendingRead == nil, "STDIO input supports one consumer") + + if bufferedInput.isEmpty == false { + let chunk = bufferedInput + bufferedInput = Data() + continuation.resume(returning: chunk) + return + } + + switch lifecycle { + case .open: + pendingRead = continuation + startReadOperationIfNeeded() + case .endOfFile: + continuation.resume(returning: nil) + case .failed(let error): + continuation.resume(throwing: error) + case .stopped: + continuation.resume(throwing: CancellationError()) + } + } + + private func startReadOperationIfNeeded() { + guard + case .open = lifecycle, + pendingRead != nil, + bufferedInput.isEmpty, + isReadOperationActive == false + else { + return + } + + isReadOperationActive = true + activeOperationByteCount = 0 + channel.read( + offset: 0, + length: Self.maxReadOperationByteCount, + queue: callbackQueue + ) { [weak self] done, dispatchData, error in + self?.receiveReadResult( + done: done, + data: dispatchData.map { Data($0) }, + error: error + ) + } + } + + private func receiveReadResult( + done: Bool, + data: Data?, + error: Int32 + ) { + guard case .open = lifecycle else { return } + + let chunk = data ?? Data() + if error != 0 { + isReadOperationActive = false + finishWithError(StdioInputReadError(code: error)) + return + } + + if chunk.isEmpty == false { + activeOperationByteCount += chunk.count + precondition( + activeOperationByteCount <= Self.maxReadOperationByteCount, + "DispatchIO exceeded the requested STDIO input byte count" + ) + deliverOrBuffer(chunk) + } + + guard done else { return } + isReadOperationActive = false + + if chunk.isEmpty { + finishAtEndOfFile() + } else { + startReadOperationIfNeeded() + } + } + + private func deliverOrBuffer(_ chunk: Data) { + if let continuation = pendingRead { + pendingRead = nil + continuation.resume(returning: chunk) + } else { + bufferedInput.append(chunk) + precondition( + bufferedInput.count <= Self.maxReadOperationByteCount, + "STDIO input buffer exceeded its bounded read operation" + ) + } + } + + private func finishAtEndOfFile() { + lifecycle = .endOfFile + channel.close() + resumeTerminalReadIfPossible() + } + + private func finishWithError(_ error: StdioInputReadError) { + lifecycle = .failed(error) + bufferedInput.removeAll() + channel.close(flags: .stop) + resumeTerminalReadIfPossible() + } + + private func resumeTerminalReadIfPossible() { + guard bufferedInput.isEmpty, let continuation = pendingRead else { return } + pendingRead = nil + switch lifecycle { + case .endOfFile: + continuation.resume(returning: nil) + case .failed(let error): + continuation.resume(throwing: error) + case .stopped: + continuation.resume(throwing: CancellationError()) + case .open: + preconditionFailure("STDIO input terminal read resumed while open") + } + } +} diff --git a/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift b/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift index d37771f..bc385ce 100644 --- a/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift +++ b/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift @@ -1,5 +1,6 @@ import Foundation import Logging +import Synchronization import Testing import XcodeMCPKit import XcodeMCPProxyTestSupport @@ -34,7 +35,7 @@ struct StdioAdapterContractTests { outputPipe.fileHandleForWriting.closeFile() } - @Test func immediateStopWaitsForReadIteratorReadiness() async throws { + @Test func immediateStopCompletesInputChannelShutdown() async throws { for _ in 0..<100 { let transport = StalledStdioAdapterTransport() let inputPipe = Pipe() @@ -58,11 +59,11 @@ struct StdioAdapterContractTests { @Test func deinitCancelsReadAndEventTasksWithoutAStopTask() async throws { let transport = StalledStdioAdapterTransport() - let inputPipe = Pipe() + let inputReader = DeinitObservingStdioInputReader() let outputPipe = Pipe() var adapter: StdioAdapter? = StdioAdapter( requestTimeout: nil, - input: inputPipe.fileHandleForReading, + inputReader: inputReader, output: outputPipe.fileHandleForWriting, recipe: MCPTransportRecipe { transport }, shutdownPolicy: .live @@ -71,12 +72,9 @@ struct StdioAdapterContractTests { try await adapter?.start() adapter = nil - for _ in 0..<100 where weakAdapter != nil { - await Task.yield() - } + await inputReader.waitUntilStopped() #expect(weakAdapter == nil) - inputPipe.fileHandleForWriting.closeFile() outputPipe.fileHandleForWriting.closeFile() } @@ -87,6 +85,46 @@ struct StdioAdapterContractTests { } } + @Test func inputReadFailureAbortsPendingRequestWithoutDrainSleep() async throws { + let payload = Data( + (initializeRequest + "\n" + initializedNotification + "\n" + toolsListRequest + "\n") + .utf8 + ) + let inputReader = ControlledFailingStdioInputReader(payload: payload) + let client = StalledStdioAdapterTransport() + let outputPipe = Pipe() + let uptimeClock = TestUptimeClock() + let drainSleepCount = Mutex(0) + let clock = ClockClient( + now: { + Date(timeIntervalSince1970: Double(uptimeClock.now()) / 1_000_000_000) + }, + uptimeNanoseconds: uptimeClock.now, + sleep: { duration in + drainSleepCount.withLock { $0 += 1 } + uptimeClock.advance(by: duration) + }, + sleepForTimeInterval: { _ in } + ) + let adapter = StdioAdapter( + requestTimeout: nil, + inputReader: inputReader, + output: outputPipe.fileHandleForWriting, + recipe: MCPTransportRecipe { client }, + shutdownPolicy: StdioAdapterShutdownPolicy(clock: clock) + ) + + try await adapter.start() + #expect(try await client.nextSentBody(at: 2) == Data(toolsListRequest.utf8)) + inputReader.fail() + await adapter.waitUntilStopped() + + #expect(drainSleepCount.withLock { $0 } == 0) + #expect(await client.sendCancellationCallCount() == 1) + #expect(await client.closeCallCount() == 1) + try outputPipe.fileHandleForWriting.close() + } + private func runEOFWithInFlightStalledRequestClosesAndCancelsClientAfterDrainTimeout() async throws { @@ -449,6 +487,10 @@ private actor StalledStdioAdapterTransport: XcodeMCPTransport { } } + func nextSentBody(at index: Int = 0) async throws -> Data { + try await sendBodies.nextValue(at: index) + } + func closeCall(at index: Int = 0) async throws { try await waitWithTimeout("waiting for fake client close") { _ = try await self.closeCalls.nextValue(at: index) @@ -469,6 +511,10 @@ private actor StalledStdioAdapterTransport: XcodeMCPTransport { await closeCalls.count() } + func sendCancellationCallCount() async -> Int { + await sendCancellationCalls.count() + } + func releaseStalledSends() async { await stalledSends.releaseAll() } @@ -516,6 +562,72 @@ private actor StalledSendContinuations { private struct StalledSendReleaseError: Error {} +private final class DeinitObservingStdioInputReader: StdioInputReading, Sendable { + private let terminal = AsyncTerminalSignal() + + func read() async throws -> Data? { + await terminal.wait() + throw CancellationError() + } + + func stop() { + terminal.signal() + } + + func waitUntilStopped() async { + await terminal.wait() + } +} + +private final class ControlledFailingStdioInputReader: StdioInputReading, Sendable { + private struct State: Sendable { + var didReadPayload = false + var isStopped = false + } + + private let payload: Data + private let state = Mutex(State()) + private let failure = AsyncTerminalSignal() + private let terminal = AsyncTerminalSignal() + + init(payload: Data) { + self.payload = payload + } + + func read() async throws -> Data? { + let shouldReturnPayload = state.withLock { state in + guard state.didReadPayload == false else { return false } + state.didReadPayload = true + return true + } + if shouldReturnPayload { + return payload + } + + await failure.wait() + if state.withLock({ $0.isStopped }) { + throw CancellationError() + } + throw ControlledStdioInputReadFailure() + } + + func fail() { + failure.signal() + } + + func stop() { + state.withLock { $0.isStopped = true } + failure.signal() + terminal.signal() + } + + func waitUntilStopped() async { + await terminal.wait() + } +} + +private struct ControlledStdioInputReadFailure: Error {} + private actor RecordedCompletionCount { private var count = 0 diff --git a/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift b/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift new file mode 100644 index 0000000..a70a6d6 --- /dev/null +++ b/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift @@ -0,0 +1,86 @@ +import Dispatch +import Foundation +import Testing +@testable import XcodeMCPProxyKit + +struct StdioInputChannelTests { + @Test func readsAcrossBoundedOperationsInOrderThenReportsEOF() async throws { + try await expectRoundTrip(byteCount: 2 * 64 * 1024 + 257) + } + + @Test func readsAnExactOperationBoundaryThenReportsEOF() async throws { + try await expectRoundTrip(byteCount: 64 * 1024) + } + + private func expectRoundTrip(byteCount: Int) async throws { + let pipe = Pipe() + let channel = StdioInputChannel(handle: pipe.fileHandleForReading) + let expected = Data( + (0..) in + DispatchQueue(label: "StdioInputChannelTests.writer").async { + do { + try pipe.fileHandleForWriting.write(contentsOf: expected) + try pipe.fileHandleForWriting.close() + continuation.resume() + } catch { + continuation.resume(throwing: error) + } + } + } + } catch { + channel.stop() + _ = await readTask.result + throw error + } + + #expect(try await readTask.value == expected) + } + + @Test func stopInterruptsActiveReadAndWaitsForDescriptorCleanup() async throws { + let pipe = Pipe() + let channel = StdioInputChannel(handle: pipe.fileHandleForReading) + let firstRead = Task { + try await channel.read() + } + let firstChunk = Data("a".utf8) + try pipe.fileHandleForWriting.write(contentsOf: firstChunk) + #expect(try await firstRead.value == firstChunk) + + let pendingRead = Task { + try await channel.read() + } + + channel.stop() + await #expect(throws: CancellationError.self) { + try await pendingRead.value + } + await channel.waitUntilStopped() + try pipe.fileHandleForWriting.close() + } + + @Test func stopBeforeReadIsIdempotent() async throws { + let pipe = Pipe() + let channel = StdioInputChannel(handle: pipe.fileHandleForReading) + + channel.stop() + channel.stop() + await channel.waitUntilStopped() + await #expect(throws: CancellationError.self) { + try await channel.read() + } + try pipe.fileHandleForWriting.close() + } +} From 3b41ed25e94af5333b14c89352d86aca728e4060 Mon Sep 17 00:00:00 2001 From: Kazuki Nakashima <65545348+lynnswap@users.noreply.github.com> Date: Wed, 15 Jul 2026 19:15:56 +0900 Subject: [PATCH 2/2] fix(stdio): preserve caller input ownership Close only the duplicated input descriptor during channel cleanup and verify that the caller-owned FileHandle remains open. Bound the failing-input test's send observation so regressions fail instead of hanging CI. --- Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift | 1 - .../StdioAdapterIntegrationTests.swift | 6 +----- Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift | 8 +++++++- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift b/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift index f8b3e65..eddee4f 100644 --- a/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift +++ b/Sources/XcodeMCPProxyKit/Stdio/StdioInputChannel.swift @@ -50,7 +50,6 @@ final class StdioInputChannel: StdioInputReading, @unchecked Sendable { queue: callbackQueue ) { _ in _ = Darwin.close(descriptor) - try? handle.close() terminal.signal() } channel.setLimit(lowWater: 1) diff --git a/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift b/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift index bc385ce..5a1e690 100644 --- a/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift +++ b/Tests/ProxyStdioAdapterTests/StdioAdapterIntegrationTests.swift @@ -115,7 +115,7 @@ struct StdioAdapterContractTests { ) try await adapter.start() - #expect(try await client.nextSentBody(at: 2) == Data(toolsListRequest.utf8)) + #expect(try await client.sentBody(at: 2) == Data(toolsListRequest.utf8)) inputReader.fail() await adapter.waitUntilStopped() @@ -487,10 +487,6 @@ private actor StalledStdioAdapterTransport: XcodeMCPTransport { } } - func nextSentBody(at index: Int = 0) async throws -> Data { - try await sendBodies.nextValue(at: index) - } - func closeCall(at index: Int = 0) async throws { try await waitWithTimeout("waiting for fake client close") { _ = try await self.closeCalls.nextValue(at: index) diff --git a/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift b/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift index a70a6d6..9e3ce14 100644 --- a/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift +++ b/Tests/ProxyStdioAdapterTests/StdioInputChannelTests.swift @@ -1,3 +1,4 @@ +import Darwin import Dispatch import Foundation import Testing @@ -47,6 +48,7 @@ struct StdioInputChannelTests { } #expect(try await readTask.value == expected) + try pipe.fileHandleForReading.close() } @Test func stopInterruptsActiveReadAndWaitsForDescriptorCleanup() async throws { @@ -68,11 +70,13 @@ struct StdioInputChannelTests { try await pendingRead.value } await channel.waitUntilStopped() + try pipe.fileHandleForReading.close() try pipe.fileHandleForWriting.close() } - @Test func stopBeforeReadIsIdempotent() async throws { + @Test func stopBeforeReadLeavesCallerHandleOpenAndIsIdempotent() async throws { let pipe = Pipe() + let callerDescriptor = pipe.fileHandleForReading.fileDescriptor let channel = StdioInputChannel(handle: pipe.fileHandleForReading) channel.stop() @@ -81,6 +85,8 @@ struct StdioInputChannelTests { await #expect(throws: CancellationError.self) { try await channel.read() } + #expect(fcntl(callerDescriptor, F_GETFD) != -1) + try pipe.fileHandleForReading.close() try pipe.fileHandleForWriting.close() } }