throw connection error if deadline is exceeded

dnadoba · dnadoba · commit 4c5733d10f6b · 2022-06-24T09:54:15.000+02:00
diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift
@@ -323,9 +323,10 @@ extension HTTPConnectionPool {
 
         mutating func cancelRequest(_ requestID: Request.ID) -> Action {
             // 1. check requests in queue
-            if self.requests.remove(requestID) != nil {
+            if let request = self.requests.remove(requestID) {
+                let error = lastConnectFailure ?? HTTPClientError.cancelled
                 return .init(
-                    request: .cancelRequestTimeout(requestID),
+                    request: .failRequest(request, error, cancelTimeout: true),
                     connection: .none
                 )
             }
diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift
@@ -444,9 +444,10 @@ extension HTTPConnectionPool {
 
         mutating func cancelRequest(_ requestID: Request.ID) -> Action {
             // 1. check requests in queue
-            if self.requests.remove(requestID) != nil {
+            if let request = self.requests.remove(requestID) {
+                let error = lastConnectFailure ?? HTTPClientError.cancelled
                 return .init(
-                    request: .cancelRequestTimeout(requestID),
+                    request: .failRequest(request, error, cancelTimeout: true),
                     connection: .none
                 )
             }
diff --git a/Sources/AsyncHTTPClient/HTTPClient.swift b/Sources/AsyncHTTPClient/HTTPClient.swift
@@ -606,7 +606,7 @@ public class HTTPClient {
             var deadlineSchedule: Scheduled<Void>?
             if let deadline = deadline {
                 deadlineSchedule = taskEL.scheduleTask(deadline: deadline) {
-                    requestBag.fail(HTTPClientError.deadlineExceeded)
+                    requestBag.cancel(.deadlineExceeded)
                 }
 
                 task.promise.futureResult.whenComplete { _ in
diff --git a/Sources/AsyncHTTPClient/HTTPHandler.swift b/Sources/AsyncHTTPClient/HTTPHandler.swift
@@ -556,7 +556,7 @@ extension URL {
 }
 
 protocol HTTPClientTaskDelegate {
-    func cancel()
+    func cancel(_ reason: CancelationReason)
 }
 
 extension HTTPClient {
@@ -618,7 +618,7 @@ extension HTTPClient {
                 return self._taskDelegate
             }
 
-            taskDelegate?.cancel()
+            taskDelegate?.cancel(.userInitiated)
         }
 
         func succeed<Delegate: HTTPClientResponseDelegate>(promise: EventLoopPromise<Response>?,
diff --git a/Sources/AsyncHTTPClient/RequestBag+StateMachine.swift b/Sources/AsyncHTTPClient/RequestBag+StateMachine.swift
@@ -31,6 +31,7 @@ extension RequestBag {
         fileprivate enum State {
             case initialized
             case queued(HTTPRequestScheduler)
+            case canceledWhileQueued(CancelationReason)
             case executing(HTTPRequestExecutor, RequestStreamState, ResponseStreamState)
             case finished(error: Error?)
             case redirected(HTTPRequestExecutor, Int, HTTPResponseHead, URL)
@@ -95,7 +96,7 @@ extension RequestBag.StateMachine {
         case .initialized, .queued:
             self.state = .executing(executor, .initialized, .initialized)
             return true
-        case .finished(error: .some):
+        case .finished(error: .some), .canceledWhileQueued:
             return false
         case .executing, .redirected, .finished(error: .none), .modifying:
             preconditionFailure("Invalid state: \(self.state)")
@@ -110,7 +111,7 @@ extension RequestBag.StateMachine {
 
     mutating func resumeRequestBodyStream() -> ResumeProducingAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("A request stream can only be resumed, if the request was started")
 
         case .executing(let executor, .initialized, .initialized):
@@ -150,7 +151,7 @@ extension RequestBag.StateMachine {
 
     mutating func pauseRequestBodyStream() {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("A request stream can only be paused, if the request was started")
         case .executing(let executor, let requestState, let responseState):
             switch requestState {
@@ -185,7 +186,7 @@ extension RequestBag.StateMachine {
 
     mutating func writeNextRequestPart(_ part: IOData, taskEventLoop: EventLoop) -> WriteAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("Invalid state: \(self.state)")
         case .executing(let executor, let requestState, let responseState):
             switch requestState {
@@ -231,7 +232,7 @@ extension RequestBag.StateMachine {
 
     mutating func finishRequestBodyStream(_ result: Result<Void, Error>) -> FinishAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("Invalid state: \(self.state)")
         case .executing(let executor, let requestState, let responseState):
             switch requestState {
@@ -282,7 +283,7 @@ extension RequestBag.StateMachine {
     /// - Returns: Whether the response should be forwarded to the delegate. Will be `false` if the request follows a redirect.
     mutating func receiveResponseHead(_ head: HTTPResponseHead) -> ReceiveResponseHeadAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("How can we receive a response, if the request hasn't started yet.")
         case .executing(let executor, let requestState, let responseState):
             guard case .initialized = responseState else {
@@ -328,7 +329,7 @@ extension RequestBag.StateMachine {
 
     mutating func receiveResponseBodyParts(_ buffer: CircularBuffer<ByteBuffer>) -> ReceiveResponseBodyAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("How can we receive a response body part, if the request hasn't started yet.")
         case .executing(_, _, .initialized):
             preconditionFailure("If we receive a response body, we must have received a head before")
@@ -385,7 +386,7 @@ extension RequestBag.StateMachine {
 
     mutating func succeedRequest(_ newChunks: CircularBuffer<ByteBuffer>?) -> ReceiveResponseEndAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("How can we receive a response body part, if the request hasn't started yet.")
         case .executing(_, _, .initialized):
             preconditionFailure("If we receive a response body, we must have received a head before")
@@ -447,7 +448,7 @@ extension RequestBag.StateMachine {
 
     private mutating func failWithConsumptionError(_ error: Error) -> ConsumeAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("Invalid state: \(self.state)")
         case .executing(_, _, .initialized):
             preconditionFailure("Invalid state: Must have received response head, before this method is called for the first time")
@@ -482,7 +483,7 @@ extension RequestBag.StateMachine {
 
     private mutating func consumeMoreBodyData() -> ConsumeAction {
         switch self.state {
-        case .initialized, .queued:
+        case .initialized, .queued, .canceledWhileQueued:
             preconditionFailure("Invalid state: \(self.state)")
 
         case .executing(_, _, .initialized):
@@ -531,9 +532,24 @@ extension RequestBag.StateMachine {
             preconditionFailure()
         }
     }
+    
+    enum CancelAction {
+        case cancelScheduler(HTTPRequestScheduler?)
+        case fail(FailAction)
+    }
+
+    mutating func cancel(_ reason: CancelationReason) -> CancelAction {
+        switch self.state {
+        case .queued(let queuer) where reason == .deadlineExceeded:
+            self.state = .canceledWhileQueued(reason)
+            return .cancelScheduler(queuer)
+        default:
+            return .fail(self.fail(reason.error))
+        }
+    }
 
     enum FailAction {
-        case failTask(HTTPRequestScheduler?, HTTPRequestExecutor?)
+        case failTask(Error, HTTPRequestScheduler?, HTTPRequestExecutor?)
         case cancelExecutor(HTTPRequestExecutor)
         case none
     }
@@ -542,31 +558,39 @@ extension RequestBag.StateMachine {
         switch self.state {
         case .initialized:
             self.state = .finished(error: error)
-            return .failTask(nil, nil)
+            return .failTask(error, nil, nil)
         case .queued(let queuer):
             self.state = .finished(error: error)
-            return .failTask(queuer, nil)
+            return .failTask(error, queuer, nil)
         case .executing(let executor, let requestState, .buffering(_, next: .eof)):
             self.state = .executing(executor, requestState, .buffering(.init(), next: .error(error)))
             return .cancelExecutor(executor)
         case .executing(let executor, _, .buffering(_, next: .askExecutorForMore)):
             self.state = .finished(error: error)
-            return .failTask(nil, executor)
+            return .failTask(error, nil, executor)
         case .executing(let executor, _, .buffering(_, next: .error(_))):
             // this would override another error, let's keep the first one
             return .cancelExecutor(executor)
         case .executing(let executor, _, .initialized):
             self.state = .finished(error: error)
-            return .failTask(nil, executor)
+            return .failTask(error, nil, executor)
         case .executing(let executor, _, .waitingForRemote):
             self.state = .finished(error: error)
-            return .failTask(nil, executor)
+            return .failTask(error, nil, executor)
         case .redirected:
             self.state = .finished(error: error)
-            return .failTask(nil, nil)
+            return .failTask(error, nil, nil)
         case .finished(.none):
             // An error occurred after the request has finished. Ignore...
             return .none
+        case .canceledWhileQueued(let reason):
+            // if we just get a `HTTPClientError.cancelled` we can use the orignal cancelation reason
+            // to give a more descriptive error to the user.
+            if (error as? HTTPClientError) == .cancelled {
+                return .failTask(reason.error, nil, nil)
+            }
+            // otherwise we already had an intermidate connection error which we should present to the user instead
+            return .failTask(error, nil, nil)
         case .finished(.some(_)):
             // this might happen, if the stream consumer has failed... let's just drop the data
             return .none
diff --git a/Sources/AsyncHTTPClient/RequestBag.swift b/Sources/AsyncHTTPClient/RequestBag.swift
@@ -320,8 +320,12 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate> {
 
         let action = self.state.fail(error)
 
+        self.executeFailAction0(action)
+    }
+    
+    private func executeFailAction0(_ action: RequestBag<Delegate>.StateMachine.FailAction) {
         switch action {
-        case .failTask(let scheduler, let executor):
+        case .failTask(let error, let scheduler, let executor):
             scheduler?.cancelRequest(self)
             executor?.cancelRequest(self)
             self.failTask0(error)
@@ -331,6 +335,33 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate> {
             break
         }
     }
+    
+    private func cancel0(_ reason: CancelationReason) {
+        self.task.eventLoop.assertInEventLoop()
+
+        let action = self.state.cancel(reason)
+
+        switch action {
+        case .cancelScheduler(let scheduler):
+            scheduler?.cancelRequest(self)
+        case .fail(let failAction):
+            self.executeFailAction0(failAction)
+        }
+    }
+}
+
+enum CancelationReason {
+    case userInitiated
+    case deadlineExceeded
+    
+    var error: HTTPClientError {
+        switch self {
+        case .userInitiated:
+            return .cancelled
+        case .deadlineExceeded:
+            return .deadlineExceeded
+        }
+    }
 }
 
 extension RequestBag: HTTPSchedulableRequest {
@@ -456,12 +487,12 @@ extension RequestBag: HTTPExecutableRequest {
 }
 
 extension RequestBag: HTTPClientTaskDelegate {
-    func cancel() {
+    func cancel(_ reason: CancelationReason) {
         if self.task.eventLoop.inEventLoop {
-            self.fail0(HTTPClientError.cancelled)
+            self.cancel0(reason)
         } else {
             self.task.eventLoop.execute {
-                self.fail0(HTTPClientError.cancelled)
+                self.cancel0(reason)
             }
         }
     }
diff --git a/Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests.swift b/Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests.swift
@@ -327,7 +327,7 @@ class HTTP1ClientChannelHandlerTests: XCTestCase {
         XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))
 
         // canceling the request
-        requestBag.cancel()
+        requestBag.cancel(.userInitiated)
         XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
             XCTAssertEqual($0 as? HTTPClientError, .cancelled)
         }
diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientRequestHandlerTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientRequestHandlerTests.swift
@@ -276,7 +276,7 @@ class HTTP2ClientRequestHandlerTests: XCTestCase {
         XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))
 
         // canceling the request
-        requestBag.cancel()
+        requestBag.cancel(.userInitiated)
         XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
             XCTAssertEqual($0 as? HTTPClientError, .cancelled)
         }
diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTests.swift b/Tests/AsyncHTTPClientTests/HTTPClientTests.swift
@@ -1268,6 +1268,47 @@ class HTTPClientTests: XCTestCase {
             #endif
         }
     }
+    
+    func testSelfSignedCertificateIsRejectedWithCorrectErrorIfRequestDeadlineIsExceeded() throws {
+        /// key + cert was created with the follwing command:
+        /// openssl req -x509 -newkey rsa:4096 -keyout self_signed_key.pem -out self_signed_cert.pem -sha256 -days 99999 -nodes -subj '/CN=localhost'
+        let certPath = Bundle.module.path(forResource: "self_signed_cert", ofType: "pem")!
+        let keyPath = Bundle.module.path(forResource: "self_signed_key", ofType: "pem")!
+        let configuration = TLSConfiguration.makeServerConfiguration(
+            certificateChain: try NIOSSLCertificate.fromPEMFile(certPath).map { .certificate($0) },
+            privateKey: .file(keyPath)
+        )
+        let sslContext = try NIOSSLContext(configuration: configuration)
+
+        let server = ServerBootstrap(group: serverGroup)
+            .childChannelInitializer { channel in
+                channel.pipeline.addHandler(NIOSSLServerHandler(context: sslContext))
+            }
+        let serverChannel = try server.bind(host: "localhost", port: 0).wait()
+        defer { XCTAssertNoThrow(try serverChannel.close().wait()) }
+        let port = serverChannel.localAddress!.port!
+
+        var config = HTTPClient.Configuration()
+        config.timeout.connect = .seconds(3)
+        let localClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup), configuration: config)
+        defer { XCTAssertNoThrow(try localClient.syncShutdown()) }
+        
+        XCTAssertThrowsError(try localClient.get(url: "https://localhost:\(port)", deadline: .now() + .seconds(2)).wait()) { error in
+            #if canImport(Network)
+            guard let nwTLSError = error as? HTTPClient.NWTLSError else {
+                XCTFail("could not cast \(error) of type \(type(of: error)) to \(HTTPClient.NWTLSError.self)")
+                return
+            }
+            XCTAssertEqual(nwTLSError.status, errSSLBadCert, "unexpected tls error: \(nwTLSError)")
+            #else
+            guard let sslError = error as? NIOSSLError,
+                  case .handshakeFailed(.sslError) = sslError else {
+                XCTFail("unexpected error \(error)")
+                return
+            }
+            #endif
+        }
+    }
 
     func testFailingConnectionIsReleased() {
         let localHTTPBin = HTTPBin(.refuse)
diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift
@@ -21,6 +21,7 @@ import XCTest
 
 class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
     func testCreatingAndFailingConnections() {
+        struct SomeError: Error, Equatable {}
         let elg = EmbeddedEventLoopGroup(loops: 4)
         defer { XCTAssertNoThrow(try elg.syncShutdownGracefully()) }
 
@@ -65,7 +66,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
 
         // fail all connection attempts
         while let randomConnectionID = connections.randomStartingConnection() {
-            struct SomeError: Error, Equatable {}
+            
 
             XCTAssertNoThrow(try connections.failConnectionCreation(randomConnectionID))
             let action = state.failedToCreateNewConnection(SomeError(), connectionID: randomConnectionID)
@@ -86,9 +87,9 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
 
         // cancel all queued requests
         while let request = queuer.timeoutRandomRequest() {
-            let cancelAction = state.cancelRequest(request)
+            let cancelAction = state.cancelRequest(request.0)
             XCTAssertEqual(cancelAction.connection, .none)
-            XCTAssertEqual(cancelAction.request, .cancelRequestTimeout(request))
+            XCTAssertEqual(cancelAction.request, .failRequest(.init(request.1), SomeError(), cancelTimeout: true))
         }
 
         // connection backoff done
@@ -184,7 +185,7 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase {
         // 2. cancel request
 
         let cancelAction = state.cancelRequest(request.id)
-        XCTAssertEqual(cancelAction.request, .cancelRequestTimeout(request.id))
+        XCTAssertEqual(cancelAction.request, .failRequest(request, HTTPClientError.cancelled, cancelTimeout: true))
         XCTAssertEqual(cancelAction.connection, .none)
 
         // 3. request timeout triggers to late
diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift
diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPoolTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPoolTests.swift
diff --git a/Tests/AsyncHTTPClientTests/Mocks/MockRequestQueuer.swift b/Tests/AsyncHTTPClientTests/Mocks/MockRequestQueuer.swift
diff --git a/Tests/AsyncHTTPClientTests/RequestBagTests.swift b/Tests/AsyncHTTPClientTests/RequestBagTests.swift

Original file line number	Diff line number	Diff line change
`@@ -323,9 +323,10 @@ extension HTTPConnectionPool {`
`323`	`323`
`324`	`324`	`mutating func cancelRequest(_ requestID: Request.ID) -> Action {`
`325`	`325`	`// 1. check requests in queue`
`326`		`- if self.requests.remove(requestID) != nil {`
	`326`	`+ if let request = self.requests.remove(requestID) {`
	`327`	`+ let error = lastConnectFailure ?? HTTPClientError.cancelled`
`327`	`328`	`return .init(`
`328`		`- request: .cancelRequestTimeout(requestID),`
	`329`	`+ request: .failRequest(request, error, cancelTimeout: true),`
`329`	`330`	`connection: .none`
`330`	`331`	`)`
`331`	`332`	`}`
Original file line number	Diff line number	Diff line change
`@@ -444,9 +444,10 @@ extension HTTPConnectionPool {`
`444`	`444`
`445`	`445`	`mutating func cancelRequest(_ requestID: Request.ID) -> Action {`
`446`	`446`	`// 1. check requests in queue`
`447`		`- if self.requests.remove(requestID) != nil {`
	`447`	`+ if let request = self.requests.remove(requestID) {`
	`448`	`+ let error = lastConnectFailure ?? HTTPClientError.cancelled`
`448`	`449`	`return .init(`
`449`		`- request: .cancelRequestTimeout(requestID),`
	`450`	`+ request: .failRequest(request, error, cancelTimeout: true),`
`450`	`451`	`connection: .none`
`451`	`452`	`)`
`452`	`453`	`}`
Original file line number	Diff line number	Diff line change
`@@ -606,7 +606,7 @@ public class HTTPClient {`
`606`	`606`	`var deadlineSchedule: Scheduled<Void>?`
`607`	`607`	`if let deadline = deadline {`
`608`	`608`	`deadlineSchedule = taskEL.scheduleTask(deadline: deadline) {`
`609`		`- requestBag.fail(HTTPClientError.deadlineExceeded)`
	`609`	`+ requestBag.cancel(.deadlineExceeded)`
`610`	`610`	`}`
`611`	`611`
`612`	`612`	`task.promise.futureResult.whenComplete { _ in`
Original file line number	Diff line number	Diff line change
`@@ -556,7 +556,7 @@ extension URL {`
`556`	`556`	`}`
`557`	`557`
`558`	`558`	`protocol HTTPClientTaskDelegate {`
`559`		`- func cancel()`
	`559`	`+ func cancel(_ reason: CancelationReason)`
`560`	`560`	`}`
`561`	`561`
`562`	`562`	`extension HTTPClient {`
`@@ -618,7 +618,7 @@ extension HTTPClient {`
`618`	`618`	`return self._taskDelegate`
`619`	`619`	`}`
`620`	`620`
`621`		`- taskDelegate?.cancel()`
	`621`	`+ taskDelegate?.cancel(.userInitiated)`
`622`	`622`	`}`
`623`	`623`
`624`	`624`	`func succeed<Delegate: HTTPClientResponseDelegate>(promise: EventLoopPromise<Response>?,`
Original file line number	Diff line number	Diff line change
`@@ -327,7 +327,7 @@ class HTTP1ClientChannelHandlerTests: XCTestCase {`
`327`	`327`	`XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))`
`328`	`328`
`329`	`329`	`// canceling the request`
`330`		`- requestBag.cancel()`
	`330`	`+ requestBag.cancel(.userInitiated)`
`331`	`331`	`XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {`
`332`	`332`	`XCTAssertEqual($0 as? HTTPClientError, .cancelled)`
`333`	`333`	`}`
Original file line number	Diff line number	Diff line change
`@@ -276,7 +276,7 @@ class HTTP2ClientRequestHandlerTests: XCTestCase {`
`276`	`276`	`XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))`
`277`	`277`
`278`	`278`	`// canceling the request`
`279`		`- requestBag.cancel()`
	`279`	`+ requestBag.cancel(.userInitiated)`
`280`	`280`	`XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {`
`281`	`281`	`XCTAssertEqual($0 as? HTTPClientError, .cancelled)`
`282`	`282`	`}`