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 <yzhuang@openai.com>

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<ByteBuffer, ByteBuffer>(
@@ -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)
+    }
+  }
+}

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