Add an AcceptBackoffHandler to the async server bootstraps (#2782)

# Motivation

When encountering a file descriptor limit we are currently closing the
server socket when using the async bootstraps since the error caught is
delivered via the `inbound` sequence of the server socket async
sequence.

# Modification

This PR adds the `AcceptBackoffHandler` in the async bootstrap case to
the pipeline to avoid us closing the server socket channel when we hit
the file descriptor limit.

# Result

We are now gracefully handling file descriptor limits.

Author: FranzBusch

Sources/NIOCore/ChannelHandlers.swift

@@ -23,6 +23,7 @@ public final class AcceptBackoffHandler: ChannelDuplexHandler, RemovableChannelH
 
     private var nextReadDeadlineNS: Optional<NIODeadline>
     private let backoffProvider: (IOError) -> TimeAmount?
+    private let shouldForwardIOErrorCaught: Bool
     private var scheduledRead: Optional<Scheduled<Void>>
 
     /// Default implementation used as `backoffProvider` which delays accept by 1 second.
@@ -38,6 +39,22 @@ public final class AcceptBackoffHandler: ChannelDuplexHandler, RemovableChannelH
         self.backoffProvider = backoffProvider
         self.nextReadDeadlineNS = nil
         self.scheduledRead = nil
+        self.shouldForwardIOErrorCaught = true
+    }
+
+    /// Create a new instance
+    ///
+    /// - parameters:
+    ///     - shouldForwardIOErrorCaught: A boolean indicating if a caught IOError should be forwarded.
+    ///     - backoffProvider: returns a `TimeAmount` which will be the amount of time to wait before attempting another `read`.
+    public init(
+        shouldForwardIOErrorCaught: Bool,
+        backoffProvider: @escaping (IOError) -> TimeAmount? = AcceptBackoffHandler.defaultBackoffProvider
+    ) {
+        self.backoffProvider = backoffProvider
+        self.nextReadDeadlineNS = nil
+        self.scheduledRead = nil
+        self.shouldForwardIOErrorCaught = shouldForwardIOErrorCaught
     }
 
     public func read(context: ChannelHandlerContext) {
@@ -59,16 +76,22 @@ public final class AcceptBackoffHandler: ChannelDuplexHandler, RemovableChannelH
     }
 
     public func errorCaught(context: ChannelHandlerContext, error: Error) {
-        if let ioError = error as? IOError {
-            if let amount = backoffProvider(ioError) {
-                self.nextReadDeadlineNS = .now() + amount
-                if let scheduled = self.scheduledRead {
-                    scheduled.cancel()
-                    scheduleRead(at: self.nextReadDeadlineNS!, context: context)
-                }
+        guard let ioError = error as? IOError else {
+            context.fireErrorCaught(error)
+            return
+        }
+
+        if let amount = backoffProvider(ioError) {
+            self.nextReadDeadlineNS = .now() + amount
+            if let scheduled = self.scheduledRead {
+                scheduled.cancel()
+                scheduleRead(at: self.nextReadDeadlineNS!, context: context)
             }
         }
-        context.fireErrorCaught(error)
+
+        if self.shouldForwardIOErrorCaught {
+            context.fireErrorCaught(error)
+        }
     }
 
     public func channelInactive(context: ChannelHandlerContext) {

Sources/NIOPosix/Bootstrap.swift

@@ -706,6 +706,10 @@ extension ServerBootstrap {
                 serverChannelInit(serverChannel)
             }.flatMap { (_) -> EventLoopFuture<NIOAsyncChannel<ChannelInitializerResult, Never>> in
                 do {
+                    try serverChannel.pipeline.syncOperations.addHandler(
+                        AcceptBackoffHandler(shouldForwardIOErrorCaught: false),
+                        name: "AcceptBackOffHandler"
+                    )
                     try serverChannel.pipeline.syncOperations.addHandler(
                         AcceptHandler(
                             childChannelInitializer: childChannelInit,

Tests/NIOPosixTests/AcceptBackoffHandlerTest.swift

@@ -14,6 +14,8 @@
 
 import Atomics
 import NIOCore
+import NIOEmbedded
+import NIOTestUtils
 import XCTest
 
 @testable import NIOPosix
@@ -300,6 +302,18 @@ public final class AcceptBackoffHandlerTest: XCTestCase {
         XCTAssertEqual(2, backoffProviderCalled.load(ordering: .relaxed))
     }
 
+    func testNotForwardingIOError() throws {
+        let loop = EmbeddedEventLoop()
+        let acceptBackOffHandler = AcceptBackoffHandler(shouldForwardIOErrorCaught: false)
+        let eventCounterHandler = EventCounterHandler()
+        let channel = EmbeddedChannel(handlers: [acceptBackOffHandler, eventCounterHandler], loop: loop)
+
+        channel.pipeline.fireErrorCaught(IOError(errnoCode: 1, reason: "test"))
+        XCTAssertEqual(eventCounterHandler.errorCaughtCalls, 0)
+        channel.pipeline.fireErrorCaught(ChannelError.alreadyClosed)
+        XCTAssertEqual(eventCounterHandler.errorCaughtCalls, 1)
+    }
+
     private final class ReadCountHandler: ChannelOutboundHandler {
         typealias OutboundIn = NIOAny
         typealias OutboundOut = NIOAny