Crash fix: HTTP2 can handle requests are cancelled

Sources/AsyncHTTPClient/ConnectionPool/HTTP2/HTTP2ClientRequestHandler.swift

@@ -193,11 +193,15 @@ final class HTTP2ClientRequestHandler: ChannelDuplexHandler {
         case .forwardResponseBodyParts(let parts):
             self.request!.receiveResponseBodyParts(parts)
 
-        case .failRequest(let error, let finalAction):
+        case .failRequest(let error, _):
             self.request!.fail(error)
             self.request = nil
             self.runTimeoutAction(.clearIdleReadTimeoutTimer, context: context)
-            self.runFinalAction(finalAction, context: context)
+            // No matter the error reason, we must always make sure the h2 stream is closed. Only
+            // once the h2 stream is closed, it is released from the h2 multiplexer. The
+            // HTTPRequestStateMachine may signal finalAction: .none in the error case (as this is
+            // the right result for HTTP/1). In the h2 case we MUST always close.
+            self.runFinalAction(.close, context: context)
 
         case .succeedRequest(let finalAction, let finalParts):
             self.request!.succeedRequest(finalParts)

Sources/AsyncHTTPClient/ConnectionPool/HTTPRequestStateMachine.swift

@@ -108,16 +108,24 @@ struct HTTPRequestStateMachine {
     }
 
     mutating func startRequest(head: HTTPRequestHead, metadata: RequestFramingMetadata) -> Action {
-        guard case .initialized = self.state else {
-            preconditionFailure("`start()` must be called first, and exactly once. Invalid state: \(self.state)")
-        }
+        switch self.state {
+        case .initialized:
+            guard self.isChannelWritable else {
+                self.state = .waitForChannelToBecomeWritable(head, metadata)
+                return .wait
+            }
+            return self.startSendingRequest(head: head, metadata: metadata)
 
-        guard self.isChannelWritable else {
-            self.state = .waitForChannelToBecomeWritable(head, metadata)
+        case .failed:
+            // The request state machine is marked as failed before the request is started, if
+            // the request was cancelled before hitting the channel handler. Before `startRequest`
+            // is called on the state machine, `willExecuteRequest` is called on
+            // `HTTPExecutableRequest`, which might loopback to state machines cancel method.
             return .wait
-        }
 
-        return self.startSendingRequest(head: head, metadata: metadata)
+        case .running, .finished, .waitForChannelToBecomeWritable, .modifying:
+            preconditionFailure("`startRequest()` must be called first, and exactly once. Invalid state: \(self.state)")
+        }
     }
 
     mutating func writabilityChanged(writable: Bool) -> Action {
@@ -381,6 +389,10 @@ struct HTTPRequestStateMachine {
         case .initialized, .waitForChannelToBecomeWritable:
             let error = HTTPClientError.cancelled
             self.state = .failed(error)
+            // Okay, this has different semantics for HTTP/1 and HTTP/2. In HTTP/1 we don't want to
+            // close the connection, if we haven't sent anything yet, to reuse the connection for
+            // another request. In HTTP/2 we must close the channel to ensure it is released from
+            // HTTP/2 multiplexer.
             return .failRequest(error, .none)
 
         case .running:

Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift

@@ -34,6 +34,7 @@ extension HTTP2ClientTests {
             ("testUncleanShutdownCancelsExecutingAndQueuedTasks", testUncleanShutdownCancelsExecutingAndQueuedTasks),
             ("testCancelingRunningRequest", testCancelingRunningRequest),
             ("testReadTimeout", testReadTimeout),
+            ("testH2CanHandleRequestsThatHaveAlreadyHitTheDeadline", testH2CanHandleRequestsThatHaveAlreadyHitTheDeadline),
             ("testStressCancelingRunningRequestFromDifferentThreads", testStressCancelingRunningRequestFromDifferentThreads),
             ("testPlatformConnectErrorIsForwardedOnTimeout", testPlatformConnectErrorIsForwardedOnTimeout),
         ]

Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift

@@ -313,6 +313,32 @@ class HTTP2ClientTests: XCTestCase {
         }
     }
 
+    func testH2CanHandleRequestsThatHaveAlreadyHitTheDeadline() {
+        let bin = HTTPBin(.http2(compress: false))
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        var config = HTTPClient.Configuration()
+        var tlsConfig = TLSConfiguration.makeClientConfiguration()
+        tlsConfig.certificateVerification = .none
+        config.tlsConfiguration = tlsConfig
+        config.httpVersion = .automatic
+        let client = HTTPClient(
+            eventLoopGroupProvider: .createNew,
+            configuration: config,
+            backgroundActivityLogger: Logger(label: "HTTPClient", factory: StreamLogHandler.standardOutput(label:))
+        )
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+
+        var request: HTTPClient.Request?
+        XCTAssertNoThrow(request = try HTTPClient.Request(url: "https://localhost:\(bin.port)"))
+
+        // just to establish an existing connection
+        XCTAssertNoThrow(try client.execute(request: XCTUnwrap(request)).wait())
+
+        XCTAssertThrowsError(try client.execute(request: XCTUnwrap(request), deadline: .now() - .seconds(2)).wait()) {
+            XCTAssertEqual($0 as? HTTPClientError, .deadlineExceeded)
+        }
+    }
+
     func testStressCancelingRunningRequestFromDifferentThreads() {
         let bin = HTTPBin(.http2(compress: false)) { _ in SendHeaderAndWaitChannelHandler() }
         defer { XCTAssertNoThrow(try bin.shutdown()) }