http2 integration tests

Author: dnadoba

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

@@ -29,6 +29,10 @@ extension HTTP2ClientTests {
             ("testConcurrentRequests", testConcurrentRequests),
             ("testConcurrentRequestsFromDifferentThreads", testConcurrentRequestsFromDifferentThreads),
             ("testConcurrentRequestsWorkWithRequiredEventLoop", testConcurrentRequestsWorkWithRequiredEventLoop),
+            ("testUncleanShutdownCancelsExecutingAndQueuedTasks", testUncleanShutdownCancelsExecutingAndQueuedTasks),
+            ("testCancelingRunningRequest", testCancelingRunningRequest),
+            ("testStressCancelingRunningRequestFromDifferentThreads", testStressCancelingRunningRequestFromDifferentThreads),
+            ("testPlatformConnectErrorIsForwardedOnTimeout", testPlatformConnectErrorIsForwardedOnTimeout),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift

@@ -37,6 +37,17 @@ class HTTP2ClientTests: XCTestCase {
         )
     }
 
+    func makeClientWithActiveHTTP2Connection<RequestHandler>(
+        to bin: HTTPBin<RequestHandler>
+    ) -> HTTPClient {
+        let client = self.makeDefaultHTTPClient()
+        var response: HTTPClient.Response?
+        XCTAssertNoThrow(response = try client.get(url: "https://localhost:\(bin.port)/get").wait())
+        XCTAssertEqual(.ok, response?.status)
+        XCTAssertEqual(response?.version, .http2)
+        return client
+    }
+
     func testSimpleGet() {
         let bin = HTTPBin(.http2(compress: false))
         defer { XCTAssertNoThrow(try bin.shutdown()) }
@@ -67,7 +78,7 @@ class HTTP2ClientTests: XCTestCase {
     func testConcurrentRequestsFromDifferentThreads() {
         let bin = HTTPBin(.http2(compress: false))
         defer { XCTAssertNoThrow(try bin.shutdown()) }
-        let client = self.makeDefaultHTTPClient()
+        let client = self.makeClientWithActiveHTTP2Connection(to: bin)
         defer { XCTAssertNoThrow(try client.syncShutdown()) }
         let numberOfWorkers = 20
         let numberOfRequestsPerWorkers = 20
@@ -92,7 +103,7 @@ class HTTP2ClientTests: XCTestCase {
 
                     for _ in 0..<numberOfRequestsPerWorkers {
                         var response: HTTPClient.Response?
-                        XCTAssertNoThrow(response = try client.get(url: url).wait())
+                        XCTAssertNoThrow(response = try client.get(url: "https://localhost:\(bin.port)/get").wait())
 
                         XCTAssertEqual(.ok, response?.status)
                         XCTAssertEqual(response?.version, .http2)
@@ -187,4 +198,167 @@ class HTTP2ClientTests: XCTestCase {
         // all workers should be running, let's wait for them to finish
         allDone.wait()
     }
+
+    func testUncleanShutdownCancelsExecutingAndQueuedTasks() {
+        let bin = HTTPBin(.http2(compress: false))
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        let client = self.makeClientWithActiveHTTP2Connection(to: bin)
+
+        // start 20 requests which are guaranteed to never get any response
+        // 10 of them will executed and the other 10 will be queued
+        // because HTTPBin has a default `maxConcurrentStreams` limit of 10
+        let responses = (0..<20).map { _ in
+            client.get(url: "https://localhost:\(bin.port)/wait")
+        }
+
+        XCTAssertNoThrow(try client.syncShutdown())
+
+        var results: [Result<HTTPClient.Response, Error>] = []
+        XCTAssertNoThrow(results = try EventLoopFuture
+            .whenAllComplete(responses, on: client.eventLoopGroup.next())
+            .timeout(after: .seconds(2))
+            .wait())
+
+        for result in results {
+            switch result {
+            case .success:
+                XCTFail("Shouldn't succeed")
+            case .failure(let error):
+                if let clientError = error as? HTTPClientError, clientError == .cancelled {
+                    continue
+                } else {
+                    XCTFail("Unexpected error: \(error)")
+                }
+            }
+        }
+    }
+
+    func testCancelingRunningRequest() {
+        let bin = HTTPBin(.http2(compress: false))
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        let client = self.makeClientWithActiveHTTP2Connection(to: bin)
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait"))
+        guard let request = maybeRequest else { return }
+
+        var task: HTTPClient.Task<TestHTTPDelegate.Response>!
+        let delegate = TestHTTPDelegate()
+        delegate.stateDidChangeCallback = { state in
+            guard case .head = state else { return }
+            // request is definitely running because we just received a head from the server
+            task.cancel()
+        }
+        task = client.execute(
+            request: request,
+            delegate: delegate
+        )
+
+        XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in
+            guard case let error = error as? HTTPClientError, error == .cancelled else {
+                return XCTFail("Should fail with cancelled")
+            }
+        }
+    }
+
+    func testStressCancelingRunningRequestFromDifferentThreads() {
+        let bin = HTTPBin(.http2(compress: false))
+        defer { XCTAssertNoThrow(try bin.shutdown()) }
+        let client = self.makeClientWithActiveHTTP2Connection(to: bin)
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+        let cancelPool = MultiThreadedEventLoopGroup(numberOfThreads: 10)
+        defer { XCTAssertNoThrow(try cancelPool.syncShutdownGracefully()) }
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait"))
+        guard let request = maybeRequest else { return }
+
+        let tasks = (0..<100).map { _ -> HTTPClient.Task<TestHTTPDelegate.Response> in
+            var task: HTTPClient.Task<TestHTTPDelegate.Response>!
+            let delegate = TestHTTPDelegate()
+            delegate.stateDidChangeCallback = { state in
+                guard case .head = state else { return }
+                // request is definitely running because we just received a head from the server
+                cancelPool.next().execute {
+                    // canceling from a different thread
+                    task.cancel()
+                }
+            }
+            task = client.execute(
+                request: request,
+                delegate: delegate
+            )
+            return task
+        }
+
+        for task in tasks {
+            XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in
+                guard case let error = error as? HTTPClientError, error == .cancelled else {
+                    return XCTFail("Should fail with cancelled")
+                }
+            }
+        }
+    }
+
+    func testPlatformConnectErrorIsForwardedOnTimeout() {
+        let bin = HTTPBin(.http2(compress: false))
+
+        let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 2)
+        let el1 = clientGroup.next()
+        let el2 = clientGroup.next()
+        defer { XCTAssertNoThrow(try clientGroup.syncShutdownGracefully()) }
+        var tlsConfig = TLSConfiguration.makeClientConfiguration()
+        tlsConfig.certificateVerification = .none
+        let client = HTTPClient(
+            eventLoopGroupProvider: .shared(clientGroup),
+            configuration: HTTPClient.Configuration(
+                tlsConfiguration: tlsConfig,
+                timeout: .init(connect: .milliseconds(1000)),
+                httpVersion: .automatic
+            ),
+            backgroundActivityLogger: Logger(label: "HTTPClient", factory: StreamLogHandler.standardOutput(label:))
+        )
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+
+        var maybeRequest1: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest1 = try HTTPClient.Request(url: "https://localhost:\(bin.port)/get"))
+        guard let request1 = maybeRequest1 else { return }
+
+        let task1 = client.execute(request: request1, delegate: ResponseAccumulator(request: request1), eventLoop: .delegateAndChannel(on: el1))
+        var response1: ResponseAccumulator.Response?
+        XCTAssertNoThrow(response1 = try task1.wait())
+
+        XCTAssertEqual(.ok, response1?.status)
+        XCTAssertEqual(response1?.version, .http2)
+        let serverPort = bin.port
+        XCTAssertNoThrow(try bin.shutdown())
+        // client is now in HTTP/2 state and the HTTPBin is closed
+        // start a new server on the old port which closes all connections immediately
+        let serverGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        defer { XCTAssertNoThrow(try serverGroup.syncShutdownGracefully()) }
+        var maybeServer: Channel?
+        XCTAssertNoThrow(maybeServer = try ServerBootstrap(group: serverGroup)
+            .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
+            .childChannelInitializer { channel in
+                channel.close()
+            }
+            .childChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
+            .bind(host: "0.0.0.0", port: serverPort)
+            .wait())
+        guard let server = maybeServer else { return }
+        defer { XCTAssertNoThrow(try server.close().wait()) }
+
+        var maybeRequest2: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest2 = try HTTPClient.Request(url: "https://localhost:\(serverPort)/"))
+        guard let request2 = maybeRequest2 else { return }
+
+        let task2 = client.execute(request: request2, delegate: ResponseAccumulator(request: request2), eventLoop: .delegateAndChannel(on: el2))
+        XCTAssertThrowsError(try task2.wait()) { error in
+            XCTAssertNil(
+                error as? HTTPClientError,
+                "error should be some platform specific error that the connection is closed/reset by the other side"
+            )
+        }
+    }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -49,11 +49,16 @@ func getDefaultEventLoopGroup(numberOfThreads: Int) -> EventLoopGroup {
 class TestHTTPDelegate: HTTPClientResponseDelegate {
     typealias Response = Void
 
-    init(backpressureEventLoop: EventLoop? = nil) {
+    init(
+        backpressureEventLoop: EventLoop? = nil,
+        stateDidChangeCallback: ((State) -> Void)? = nil
+    ) {
         self.backpressureEventLoop = backpressureEventLoop
+        self.stateDidChangeCallback = stateDidChangeCallback
     }
 
     var backpressureEventLoop: EventLoop?
+    var stateDidChangeCallback: ((State) -> Void)?
 
     enum State {
         case idle
@@ -63,7 +68,11 @@ class TestHTTPDelegate: HTTPClientResponseDelegate {
         case error(Error)
     }
 
-    var state = State.idle
+    var state = State.idle {
+        didSet {
+            self.stateDidChangeCallback?(self.state)
+        }
+    }
 
     func didReceiveHead(task: HTTPClient.Task<Response>, _ head: HTTPResponseHead) -> EventLoopFuture<Void> {
         self.state = .head(head)
@@ -804,6 +813,10 @@ internal final class HTTPBinHandler: ChannelInboundHandler {
                 builder.add(buff)
                 self.resps.append(builder)
                 return
+            case "/sendheaderandwait":
+                // sends some headers and waits indefinitely afterwards
+                context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok))), promise: nil)
+                return
             case "/wait":
                 return
             case "/close":