Add test for HTTP1 request with large header

Sources/AsyncHTTPClient/ConnectionPool/HTTP1/HTTP1ClientChannelHandler.swift

@@ -35,8 +35,8 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
         didSet {
             if let newRequest = self.request {
                 var requestLogger = newRequest.logger
-                requestLogger[metadataKey: "ahc-connection-id"] = "\(self.connection.id)"
-                requestLogger[metadataKey: "ahc-el"] = "\(self.connection.channel.eventLoop)"
+                requestLogger[metadataKey: "ahc-connection-id"] = self.connectionIdLoggerMetadata
+                requestLogger[metadataKey: "ahc-el"] = "\(self.eventLoop)"
                 self.logger = requestLogger
 
                 if let idleReadTimeout = newRequest.requestOptions.idleReadTimeout {
@@ -59,15 +59,15 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
 
     private let backgroundLogger: Logger
     private var logger: Logger
+    private let eventLoop: EventLoop
+    private let connectionIdLoggerMetadata: Logger.MetadataValue
 
-    let connection: HTTP1Connection
-    let eventLoop: EventLoop
-
-    init(connection: HTTP1Connection, eventLoop: EventLoop, logger: Logger) {
-        self.connection = connection
+    var onConnectionIdle: () -> Void = {}
+    init(eventLoop: EventLoop, backgroundLogger: Logger, connectionIdLoggerMetadata: Logger.MetadataValue) {
         self.eventLoop = eventLoop
-        self.backgroundLogger = logger
-        self.logger = self.backgroundLogger
+        self.backgroundLogger = backgroundLogger
+        self.logger = backgroundLogger
+        self.connectionIdLoggerMetadata = connectionIdLoggerMetadata
     }
 
     func handlerAdded(context: ChannelHandlerContext) {
@@ -108,6 +108,7 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
 
         let action = self.state.writabilityChanged(writable: context.channel.isWritable)
         self.run(action, context: context)
+        context.fireChannelWritabilityChanged()
     }
 
     func channelRead(context: ChannelHandlerContext, data: NIOAny) {
@@ -274,7 +275,7 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
                         if shouldClose {
                             context.close(promise: nil)
                         } else {
-                            self.connection.taskCompleted()
+                            self.onConnectionIdle()
                         }
 
                         oldRequest.succeedRequest(buffer)
@@ -286,7 +287,7 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
 
                 context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: writePromise)
             case .informConnectionIsIdle:
-                self.connection.taskCompleted()
+                self.onConnectionIdle()
                 oldRequest.succeedRequest(buffer)
             }
 
@@ -303,7 +304,7 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
                 oldRequest.fail(error)
 
             case .informConnectionIsIdle:
-                self.connection.taskCompleted()
+                self.onConnectionIdle()
                 oldRequest.fail(error)
 
             case .failWritePromise(let writePromise):
@@ -328,6 +329,7 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
             // we must check if the request is still present here.
             guard let request = self.request else { return }
             request.requestHeadSent()
+
             request.resumeRequestBodyStream()
         } else {
             context.write(self.wrapOutboundOut(.head(head)), promise: nil)
@@ -434,6 +436,11 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
     }
 }
 
+#if swift(>=5.6)
+@available(*, unavailable)
+extension HTTP1ClientChannelHandler: Sendable {}
+#endif
+
 extension HTTP1ClientChannelHandler: HTTPRequestExecutor {
     func writeRequestBodyPart(_ data: IOData, request: HTTPExecutableRequest, promise: EventLoopPromise<Void>?) {
         if self.eventLoop.inEventLoop {

Sources/AsyncHTTPClient/ConnectionPool/HTTP1/HTTP1Connection.swift

@@ -133,10 +133,13 @@ final class HTTP1Connection {
             }
 
             let channelHandler = HTTP1ClientChannelHandler(
-                connection: self,
                 eventLoop: channel.eventLoop,
-                logger: logger
+                backgroundLogger: logger,
+                connectionIdLoggerMetadata: "\(self.id)"
             )
+            channelHandler.onConnectionIdle = {
+                self.taskCompleted()
+            }
 
             try sync.addHandler(channelHandler)
         } catch {

Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests+XCTest.swift

@@ -33,6 +33,7 @@ extension HTTP1ClientChannelHandlerTests {
             ("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand),
             ("testWriteHTTPHeadFails", testWriteHTTPHeadFails),
             ("testHandlerClosesChannelIfLastActionIsSendEndAndItFails", testHandlerClosesChannelIfLastActionIsSendEndAndItFails),
+            ("testChannelBecomesNonWritableDuringHeaderWrite", testChannelBecomesNonWritableDuringHeaderWrite),
         ]
     }
 }

Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests.swift

@@ -526,6 +526,37 @@ class HTTP1ClientChannelHandlerTests: XCTestCase {
             XCTAssertTrue(error is FailEndHandler.Error)
         }
     }
+
+    func testChannelBecomesNonWritableDuringHeaderWrite() throws {
+        try XCTSkipIf(true, "this currently fails and will be fixed in follow up PR")
+        final class ChangeWritabilityOnFlush: ChannelOutboundHandler {
+            typealias OutboundIn = Any
+            func flush(context: ChannelHandlerContext) {
+                context.flush()
+                (context.channel as! EmbeddedChannel).isWritable = false
+                context.fireChannelWritabilityChanged()
+            }
+        }
+        let eventLoopGroup = EmbeddedEventLoopGroup(loops: 1)
+        let eventLoop = eventLoopGroup.next() as! EmbeddedEventLoop
+        let handler = HTTP1ClientChannelHandler(
+            eventLoop: eventLoop,
+            backgroundLogger: Logger(label: "no-op", factory: SwiftLogNoOpLogHandler.init),
+            connectionIdLoggerMetadata: "test connection"
+        )
+        let channel = EmbeddedChannel(handlers: [
+            ChangeWritabilityOnFlush(),
+            handler,
+        ], loop: eventLoop)
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 80)).wait()
+
+        let request = MockHTTPExecutableRequest()
+        // non empty body is important to trigger this bug as we otherwise finish the request in a single flush
+        request.requestFramingMetadata.body = .fixedSize(1)
+        request.raiseErrorIfUnimplementedMethodIsCalled = false
+        channel.writeAndFlush(request, promise: nil)
+        XCTAssertEqual(request.events.map(\.kind), [.willExecuteRequest, .requestHeadSent])
+    }
 }
 
 class TestBackpressureWriter {

Tests/AsyncHTTPClientTests/HTTPClientBase.swift

@@ -39,7 +39,7 @@ class XCTestCaseHTTPClientTestsBaseClass: XCTestCase {
     var backgroundLogStore: CollectEverythingLogHandler.LogStore!
 
     var defaultHTTPBinURLPrefix: String {
-        return "http://localhost:\(self.defaultHTTPBin.port)/"
+        self.defaultHTTPBin.baseURL
     }
 
     override func setUp() {

Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -375,7 +375,18 @@ internal final class HTTPBin<RequestHandler: ChannelInboundHandler> where
                 return "https"
             }
         }()
-        return "\(scheme)://localhost:\(self.port)/"
+        let host: String = {
+            switch self.socketAddress {
+            case .v4:
+                return self.socketAddress.ipAddress!
+            case .v6:
+                return "[\(self.socketAddress.ipAddress!)]"
+            case .unixDomainSocket:
+                return self.socketAddress.pathname!
+            }
+        }()
+
+        return "\(scheme)://\(host):\(self.port)/"
     }
 
     private let mode: Mode

Tests/AsyncHTTPClientTests/HTTPClientTests+XCTest.swift

@@ -143,6 +143,7 @@ extension HTTPClientTests {
             ("testRequestWithHeaderTransferEncodingIdentityDoesNotFail", testRequestWithHeaderTransferEncodingIdentityDoesNotFail),
             ("testMassiveDownload", testMassiveDownload),
             ("testShutdownWithFutures", testShutdownWithFutures),
+            ("testMassiveHeaderHTTP1", testMassiveHeaderHTTP1),
         ]
     }
 }

Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -3363,4 +3363,19 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass {
         let httpClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup))
         XCTAssertNoThrow(try httpClient.shutdown().wait())
     }
+
+    func testMassiveHeaderHTTP1() throws {
+        try XCTSkipIf(true, "this currently crashes and will be fixed in follow up PR")
+        var request = try HTTPClient.Request(url: defaultHTTPBin.baseURL, method: .POST)
+        // add ~64 KB header
+        let headerValue = String(repeating: "0", count: 1024)
+        for headerID in 0..<64 {
+            request.headers.replaceOrAdd(name: "larg-header-\(headerID)", value: headerValue)
+        }
+
+        // non empty body is important to trigger this bug as we otherwise finish the request in a single flush
+        request.body = .byteBuffer(ByteBuffer(bytes: [0]))
+
+        XCTAssertNoThrow(try defaultClient.execute(request: request).wait())
+    }
 }

Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift

@@ -37,7 +37,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         // for the first eight requests, the pool should try to create new connections.
 
         for _ in 0..<8 {
-            let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
             guard case .createConnection(let connectionID, let connectionEL) = action.connection else {
@@ -53,7 +53,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         // the next eight requests should only be queued.
 
         for _ in 0..<8 {
-            let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
             guard case .none = action.connection else {
@@ -120,7 +120,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         // for the first eight requests, the pool should try to create new connections.
 
         for _ in 0..<8 {
-            let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
             guard case .createConnection(let connectionID, let connectionEL) = action.connection else {
@@ -136,7 +136,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         // the next eight requests should only be queued.
 
         for _ in 0..<8 {
-            let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
             guard case .none = action.connection else {
@@ -181,7 +181,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let action = state.executeRequest(request)
@@ -239,7 +239,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let executeAction = state.executeRequest(request)
@@ -276,7 +276,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let executeAction = state.executeRequest(request)
@@ -310,7 +310,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         XCTAssertEqual(cleanupContext.connectBackoff, [])
 
         // 4. execute another request
-        let finalMockRequest = MockHTTPRequest(eventLoop: elg.next())
+        let finalMockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
         let finalRequest = HTTPConnectionPool.Request(finalMockRequest)
         let failAction = state.executeRequest(finalRequest)
         XCTAssertEqual(failAction.connection, .none)
@@ -339,7 +339,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
                 return XCTFail("Expected to still have connections available")
             }
 
-            let mockRequest = MockHTTPRequest(eventLoop: eventLoop)
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: eventLoop)
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
 
@@ -359,7 +359,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         var queuer = MockRequestQueuer()
         for _ in 0..<100 {
             let eventLoop = elg.next()
-            let mockRequest = MockHTTPRequest(eventLoop: eventLoop, requiresEventLoopForChannel: false)
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: eventLoop, requiresEventLoopForChannel: false)
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
 
@@ -418,7 +418,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
 
             // 10% of the cases enforce the eventLoop
             let elRequired = (0..<10).randomElement().flatMap { $0 == 0 ? true : false }!
-            let mockRequest = MockHTTPRequest(eventLoop: reqEventLoop, requiresEventLoopForChannel: elRequired)
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: reqEventLoop, requiresEventLoopForChannel: elRequired)
             let request = HTTPConnectionPool.Request(mockRequest)
 
             let action = state.executeRequest(request)
@@ -482,7 +482,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         XCTAssertEqual(connections.parked, 8)
 
         // close a leased connection == abort
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next())
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next())
         let request = HTTPConnectionPool.Request(mockRequest)
         guard let connectionToAbort = connections.newestParkedConnection else {
             return XCTFail("Expected to have a parked connection")
@@ -536,7 +536,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
                 return XCTFail("Expected to still have connections available")
             }
 
-            let mockRequest = MockHTTPRequest(eventLoop: eventLoop)
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: eventLoop)
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
 
@@ -553,7 +553,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         for _ in 0..<100 {
             let eventLoop = elg.next()
 
-            let mockRequest = MockHTTPRequest(eventLoop: eventLoop, requiresEventLoopForChannel: false)
+            let mockRequest = MockHTTPScheduableRequest(eventLoop: eventLoop, requiresEventLoopForChannel: false)
             let request = HTTPConnectionPool.Request(mockRequest)
             let action = state.executeRequest(request)
 
@@ -667,7 +667,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let executeAction = state.executeRequest(request)
@@ -706,7 +706,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let executeAction = state.executeRequest(request)
@@ -738,7 +738,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest = MockHTTPRequest(eventLoop: eventLoop.next(), requiresEventLoopForChannel: false)
+        let mockRequest = MockHTTPScheduableRequest(eventLoop: eventLoop.next(), requiresEventLoopForChannel: false)
         let request = HTTPConnectionPool.Request(mockRequest)
 
         let executeAction = state.executeRequest(request)
@@ -762,7 +762,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
             retryConnectionEstablishment: true
         )
 
-        let mockRequest1 = MockHTTPRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
+        let mockRequest1 = MockHTTPScheduableRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
         let request1 = HTTPConnectionPool.Request(mockRequest1)
 
         let executeAction1 = state.executeRequest(request1)
@@ -773,7 +773,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
 
         XCTAssertEqual(executeAction1.request, .scheduleRequestTimeout(for: request1, on: mockRequest1.eventLoop))
 
-        let mockRequest2 = MockHTTPRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
+        let mockRequest2 = MockHTTPScheduableRequest(eventLoop: elg.next(), requiresEventLoopForChannel: false)
         let request2 = HTTPConnectionPool.Request(mockRequest2)
 
         let executeAction2 = state.executeRequest(request2)