From e3b4f71f776d924eeef5476acf30841833c94df8 Mon Sep 17 00:00:00 2001 From: RKS Date: Mon, 28 Sep 2026 16:22:06 -0400 Subject: [PATCH] fix(control-socket): recover from transient accept errors (#1351) * fix(control-socket): recover from transient accept errors Keep the control socket alive when Darwin reports a transient fcntl failure while accepting a client connection. Refs #1346 * test(control-socket): use synchronous pipeline lookup * test(control-socket): query handler on event loop * fix(control-socket): preserve channel backpressure * test(control-socket): exercise accept error recovery * test(control-socket): verify accepts continue after transient errors Check that the real server inbound stream yields a client connection after each simulated accept error. Keep the read-routing and unrelated-error checks, bound the wait, and close the test connections. Adapted from the regression test suggested in: https://github.com/openai/tart/pull/1351#issuecomment-5851285933 --------- Co-authored-by: Yibo Zhuang --- Sources/tart/ControlSocket.swift | 19 +++++ Tests/TartTests/ControlSocketTests.swift | 100 +++++++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/Sources/tart/ControlSocket.swift b/Sources/tart/ControlSocket.swift index 076928c..680d515 100644 --- a/Sources/tart/ControlSocket.swift +++ b/Sources/tart/ControlSocket.swift @@ -37,6 +37,12 @@ class ControlSocket { do { self.serverChannel = try await ServerBootstrap(group: eventLoopGroup) + .serverChannelInitializer { channel in + channel.pipeline.addHandler( + ControlSocketAcceptErrorHandler(), + name: "ControlSocketAcceptErrorHandler" + ) + } .bind(unixDomainSocketPath: controlSocketURL.relativePath) { childChannel in childChannel.eventLoop.makeCompletedFuture { return try NIOAsyncChannel( @@ -125,3 +131,16 @@ class ControlSocket { return fd } } + +private final class ControlSocketAcceptErrorHandler: ChannelInboundHandler { + typealias InboundIn = Channel + typealias InboundOut = Channel + + func errorCaught(context: ChannelHandlerContext, error: Error) { + if error is NIOFcntlFailedError { + context.channel.read() + } else { + context.fireErrorCaught(error) + } + } +} diff --git a/Tests/TartTests/ControlSocketTests.swift b/Tests/TartTests/ControlSocketTests.swift index fb896ef..88d3bd7 100644 --- a/Tests/TartTests/ControlSocketTests.swift +++ b/Tests/TartTests/ControlSocketTests.swift @@ -1,5 +1,6 @@ import NIO import XCTest +@testable import NIOPosix @testable import tart // Avoid NSObject.bind and Tart's Darwin type shadowing the system function. @@ -30,6 +31,88 @@ final class ControlSocketTests: XCTestCase { try await eventLoopGroup.shutdownGracefully() } + func testAcceptErrorsRetryReadingAndForwardOtherErrors() async throws { + let temporaryDirectory = try makeTemporaryDirectory() + let originalDirectory = FileManager.default.currentDirectoryPath + defer { + FileManager.default.changeCurrentDirectoryPath(originalDirectory) + try? FileManager.default.removeItem(at: temporaryDirectory) + } + + let socketURL = URL(fileURLWithPath: "control.sock", relativeTo: temporaryDirectory) + let controlSocket = try await ControlSocket(socketURL) + let channel = controlSocket.serverChannel.channel + let observations = try await channel.eventLoop.submit { + let observer = AcceptErrorObserver() + // Placing this after the recovery handler also detects context.read(), which + // bypasses downstream backpressure handlers instead of starting at the tail. + try channel.pipeline.syncOperations.addHandler(observer) + for _ in 0..<3 { + channel.pipeline.fireErrorCaught(NIOFcntlFailedError()) + } + let retries = observer.readCount + let transientErrors = observer.errors.count + channel.pipeline.fireErrorCaught(ChannelError.inputClosed) + return (retries, transientErrors, observer.readCount, + observer.errors.count, observer.errors.first as? ChannelError) + }.get() + + try await controlSocket.serverChannel.executeThenClose { _ in } + try await controlSocket.eventLoopGroup.shutdownGracefully() + + XCTAssertEqual(observations.0, 3) + XCTAssertEqual(observations.1, 0) + XCTAssertEqual(observations.2, 3) + XCTAssertEqual(observations.3, 1) + XCTAssertEqual(observations.4, .inputClosed) + } + + func testAcceptErrorsDoNotEndInboundConnections() async throws { + let temporaryDirectory = try makeTemporaryDirectory() + let originalDirectory = FileManager.default.currentDirectoryPath + defer { + FileManager.default.changeCurrentDirectoryPath(originalDirectory) + try? FileManager.default.removeItem(at: temporaryDirectory) + } + + let socketURL = URL(fileURLWithPath: "control.sock", relativeTo: temporaryDirectory) + let controlSocket = try await ControlSocket(socketURL) + let serverChannel = controlSocket.serverChannel + + do { + try await serverChannel.executeThenClose { inbound in + // Bound the wait if the listener stays open but stops accepting connections. + let timeout = serverChannel.channel.eventLoop.scheduleTask(in: .seconds(10)) { + serverChannel.channel.close(promise: nil) + } + defer { timeout.cancel() } + + var iterator = inbound.makeAsyncIterator() + for _ in 0..<3 { + try await serverChannel.channel.eventLoop.submit { + serverChannel.channel.pipeline.fireErrorCaught(NIOFcntlFailedError()) + }.get() + + let clientChannel = try await ClientBootstrap(group: controlSocket.eventLoopGroup) + .connectTimeout(.seconds(5)) + .connect(unixDomainSocketPath: socketURL.path) + .get() + defer { clientChannel.close(promise: nil) } + + // ControlSocket.run() consumes this stream. A recoverable accept error + // must not prevent it from receiving the next connection. + let nextChannel = try await iterator.next() + let acceptedChannel = try XCTUnwrap(nextChannel, "The listener stopped delivering connections after an accept error") + try await acceptedChannel.executeThenClose { _, _ in } + } + } + } catch { + try? await controlSocket.eventLoopGroup.shutdownGracefully() + throw error + } + try await controlSocket.eventLoopGroup.shutdownGracefully() + } + func testInitializerPropagatesControlSocketCreationFailure() async throws { let temporaryDirectory = try makeTemporaryDirectory() let originalDirectory = FileManager.default.currentDirectoryPath @@ -94,3 +177,20 @@ final class ControlSocketTests: XCTestCase { return directory } } + +private final class AcceptErrorObserver: ChannelDuplexHandler { + typealias InboundIn = Channel + typealias OutboundIn = ByteBuffer + + var readCount = 0 + var errors: [Error] = [] + + func read(context: ChannelHandlerContext) { + readCount += 1 + context.read() + } + + func errorCaught(context: ChannelHandlerContext, error: Error) { + errors.append(error) + } +}