Add defensive connection closure.

Motivation:

Currently when either we or the server send Connection: close, we
correctly do not return that connection to the pool. However, we rely on
the server actually performing the connection closure: we never call
close() ourselves. This is unnecessarily optimistic: a server may
absolutely fail to close this connection. To protect our own file
descriptors, we should make sure that any connection we do not return
the pool is closed.

Modifications:

If we think a connection is closing when we release it, we now call
close() on it defensively.

Result:

We no longer leak connections when the server fails to close them.

Fixes #324.

Author: Lukasa

Diff for: Sources/AsyncHTTPClient/Connection.swift

@@ -72,6 +72,10 @@ extension Connection {
     func close() -> EventLoopFuture<Void> {
         return self.channel.close()
     }
+
+    func close(promise: EventLoopPromise<Void>?) {
+        return self.channel.close(promise: promise)
+    }
 }
 
 /// Methods of Connection which are used in ConnectionsState extracted as protocol

Diff for: Sources/AsyncHTTPClient/ConnectionPool.swift

@@ -352,6 +352,15 @@ class HTTP1ConnectionProvider {
             self.state.release(connection: connection, closing: closing)
         }
 
+        // We close defensively here: we may have failed to actually close on other codepaths,
+        // or we may be expecting the server to close. In either case, we want our FD back, so
+        // we close now to cover our backs. We don't care about the result: if the channel is
+        // _already_ closed, that's fine by us.
+        if closing {
+            logger.debug("closing connection")
+            connection.close(promise: nil)
+        }
+
         switch action {
         case .none:
             break

Diff for: Tests/AsyncHTTPClientTests/HTTPClientInternalTests.swift

@@ -812,7 +812,7 @@ class HTTPClientInternalTests: XCTestCase {
 
         XCTAssert(connection !== connection2)
         try! connection2.channel.eventLoop.submit {
-            connection2.release(closing: true, logger: HTTPClient.loggingDisabled)
+            connection2.release(closing: false, logger: HTTPClient.loggingDisabled)
         }.wait()
         XCTAssertTrue(connection2.channel.isActive)
     }

Diff for: Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -772,6 +772,53 @@ internal final class HttpBinForSSLUncleanShutdownHandler: ChannelInboundHandler
     }
 }
 
+internal final class CloseWithoutClosingServerHandler: ChannelInboundHandler {
+    typealias InboundIn = HTTPServerRequestPart
+    typealias OutboundOut = HTTPServerResponsePart
+
+    private var callback: (() -> Void)?
+    private var onClosePromise: EventLoopPromise<Void>?
+
+    init(_ callback: @escaping () -> Void) {
+        self.callback = callback
+    }
+
+    func handlerAdded(context: ChannelHandlerContext) {
+        self.onClosePromise = context.eventLoop.makePromise()
+        self.onClosePromise!.futureResult.whenSuccess(self.callback!)
+        self.callback = nil
+    }
+
+    func handlerRemoved(context: ChannelHandlerContext) {
+        assert(self.onClosePromise == nil)
+    }
+
+    func channelInactive(context: ChannelHandlerContext) {
+        if let onClosePromise = self.onClosePromise {
+            self.onClosePromise = nil
+            onClosePromise.succeed(())
+        }
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        guard case .end = self.unwrapInboundIn(data) else {
+            return
+        }
+
+        // We're gonna send a response back here, with Connection: close, but we will
+        // not close the connection. This reproduces #324.
+        let headers = HTTPHeaders([
+            ("Host", "CloseWithoutClosingServerHandler"),
+            ("Content-Length", "0"),
+            ("Connection", "close"),
+        ])
+        let head = HTTPResponseHead(version: .init(major: 1, minor: 1), status: .ok, headers: headers)
+
+        context.write(self.wrapOutboundOut(.head(head)), promise: nil)
+        context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
+    }
+}
+
 struct EventLoopFutureTimeoutError: Error {}
 
 extension EventLoopFuture {

Diff for: Tests/AsyncHTTPClientTests/HTTPClientTests+XCTest.swift

@@ -126,6 +126,7 @@ extension HTTPClientTests {
             ("testNoBytesSentOverBodyLimit", testNoBytesSentOverBodyLimit),
             ("testDoubleError", testDoubleError),
             ("testSSLHandshakeErrorPropagation", testSSLHandshakeErrorPropagation),
+            ("testWeCloseConnectionsWhenConnectionCloseSetByServer", testWeCloseConnectionsWhenConnectionCloseSetByServer),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -2712,4 +2712,30 @@ class HTTPClientTests: XCTestCase {
             #endif
         }
     }
+
+    func testWeCloseConnectionsWhenConnectionCloseSetByServer() throws {
+        let group = DispatchGroup()
+        group.enter()
+
+        let server = try ServerBootstrap(group: self.serverGroup)
+            .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
+            .childChannelInitializer { channel in
+                channel.pipeline.configureHTTPServerPipeline().flatMap {
+                    channel.pipeline.addHandler(CloseWithoutClosingServerHandler(group.leave))
+                }
+            }
+            .bind(host: "localhost", port: 0)
+            .wait()
+
+        defer {
+            server.close(promise: nil)
+        }
+
+        // Simple request, should go great.
+        XCTAssertNoThrow(try self.defaultClient.get(url: "http://localhost:\(server.localAddress!.port!)/").wait())
+
+        // Shouldn't need more than 100ms of waiting to see the close.
+        let result = group.wait(timeout: .now().advanced(by: .milliseconds(100)))
+        XCTAssertEqual(result, .success, "we never closed the connection!")
+    }
 }