New test

Author: fabianfett

Sources/AsyncHTTPClient/ConnectionPool/HTTPExecutingRequest.swift

@@ -226,15 +226,19 @@ protocol HTTPExecutingRequest: AnyObject {
     /// to ask for more data.
     func receiveResponseHead(_ head: HTTPResponseHead)
 
-    /// Receive a response body stream part.
+    /// Receive response body stream parts.
     ///
     /// Please note that `receiveResponseHead` and `receiveResponseBodyPart` may
     /// be called in quick succession. It is the task's job to buffer those events for the user. Once all
     /// buffered data has been consumed the task must call `executor.demandResponseBodyStream`
     /// to ask for more data.
     func receiveResponseBodyParts(_ buffer: CircularBuffer<ByteBuffer>)
 
+    /// Succeeds the executing request. The executor will not call any further methods on the request after this method.
+    ///
+    /// - Parameter buffer: The remaining response body parts, that were received before the request end
     func succeedRequest(_ buffer: CircularBuffer<ByteBuffer>?)
 
+    /// Fails the executing request, with an error.
     func fail(_ error: Error)
 }

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -923,6 +923,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
         case httpProxyHandshakeTimeout
         case tlsHandshakeTimeout
         case serverOfferedUnsupportedApplicationProtocol(String)
+        case requestStreamCancelled
     }
 
     private var code: Code
@@ -991,4 +992,8 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     public static func serverOfferedUnsupportedApplicationProtocol(_ proto: String) -> HTTPClientError {
         return HTTPClientError(code: .serverOfferedUnsupportedApplicationProtocol(proto))
     }
+
+    /// The remote server responded with a status code between >= 300, before the full request was send. The request stream
+    /// was therefore cancelled
+    public static let requestStreamCancelled = HTTPClientError(code: .requestStreamCancelled)
 }

Sources/AsyncHTTPClient/RequestBag+StateMachine.swift

@@ -203,7 +203,7 @@ extension RequestBag.StateMachine {
         case .finished(error: .some(let error)):
             return .failFuture(error)
         case .finished(error: .none):
-            preconditionFailure("A write was made, after the request has completed")
+            return .failFuture(HTTPClientError.requestStreamCancelled)
         case .modifying:
             preconditionFailure("Invalid state")
         }
@@ -338,7 +338,10 @@ extension RequestBag.StateMachine {
                 preconditionFailure("If we have received an error or eof before, why did we get another body part? Next: \(next)")
             }
 
-            if buffer.isEmpty, let newChunks = newChunks {
+            if buffer.isEmpty, newChunks == nil || newChunks!.isEmpty {
+                self.state = .finished(error: nil)
+                return .succeedRequest
+            } else if buffer.isEmpty, let newChunks = newChunks {
                 buffer = newChunks
             } else if let newChunks = newChunks {
                 buffer.append(contentsOf: newChunks)

Tests/AsyncHTTPClientTests/RequestBagTests+XCTest.swift

@@ -30,6 +30,7 @@ extension RequestBagTests {
             ("testCancelFailsTaskBeforeRequestIsSent", testCancelFailsTaskBeforeRequestIsSent),
             ("testCancelFailsTaskAfterRequestIsSent", testCancelFailsTaskAfterRequestIsSent),
             ("testCancelFailsTaskWhenTaskIsQueued", testCancelFailsTaskWhenTaskIsQueued),
+            ("testHTTPUploadIsCancelledEvenThoughRequestSucceeds", testHTTPUploadIsCancelledEvenThoughRequestSucceeds),
         ]
     }
 }

Tests/AsyncHTTPClientTests/RequestBagTests.swift

@@ -66,7 +66,7 @@ final class RequestBagTests: XCTestCase {
         )
         XCTAssert(bag.task.eventLoop === embeddedEventLoop)
 
-        let executor = MockRequestExecutor()
+        let executor = MockRequestExecutor(pauseRequestBodyPartStreamAfterASingleWrite: true)
 
         bag.willExecuteRequest(executor)
 
@@ -294,6 +294,71 @@ final class RequestBagTests: XCTestCase {
             XCTAssertEqual($0 as? HTTPClientError, .cancelled)
         }
     }
+
+    func testHTTPUploadIsCancelledEvenThoughRequestSucceeds() {
+        let embeddedEventLoop = EmbeddedEventLoop()
+        defer { XCTAssertNoThrow(try embeddedEventLoop.syncShutdownGracefully()) }
+        let logger = Logger(label: "test")
+
+        var maybeRequest: HTTPClient.Request?
+        let writeSecondPartPromise = embeddedEventLoop.makePromise(of: Void.self)
+
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(
+            url: "https://swift.org",
+            method: .POST,
+            headers: ["content-length": "12"],
+            body: .stream(length: 12) { writer -> EventLoopFuture<Void> in
+                var firstWriteSuccess = false
+                return writer.write(.byteBuffer(.init(bytes: 0...3))).flatMap { _ in
+                    firstWriteSuccess = true
+
+                    return writeSecondPartPromise.futureResult
+                }.flatMap {
+                    return writer.write(.byteBuffer(.init(bytes: 4...7)))
+                }.always { result in
+                    XCTAssertTrue(firstWriteSuccess)
+
+                    guard case .failure(let error) = result else {
+                        return XCTFail("Expected the second write to fail")
+                    }
+                    XCTAssertEqual(error as? HTTPClientError, .requestStreamCancelled)
+                }
+            }
+        ))
+        guard let request = maybeRequest else { return XCTFail("Expected to have a request") }
+
+        let delegate = UploadCountingDelegate(eventLoop: embeddedEventLoop)
+        let bag = RequestBag(
+            request: request,
+            eventLoopPreference: .delegate(on: embeddedEventLoop),
+            task: .init(eventLoop: embeddedEventLoop, logger: logger),
+            redirectHandler: nil,
+            connectionDeadline: .now() + .seconds(30),
+            idleReadTimeout: nil,
+            delegate: delegate
+        )
+
+        let executor = MockRequestExecutor()
+        bag.willExecuteRequest(executor)
+
+        XCTAssertEqual(delegate.hitDidSendRequestHead, 0)
+        XCTAssertEqual(delegate.hitDidSendRequest, 0)
+        bag.requestHeadSent()
+        XCTAssertEqual(delegate.hitDidSendRequestHead, 1)
+        XCTAssertEqual(delegate.hitDidSendRequest, 0)
+
+        bag.resumeRequestBodyStream()
+        XCTAssertEqual(executor.nextBodyPart(), .body(.byteBuffer(.init(bytes: 0...3))))
+        // receive a 301 response immediately.
+        bag.receiveResponseHead(.init(version: .http1_1, status: .movedPermanently))
+        bag.succeedRequest(.init())
+
+        // if we now write our second part of the response this should fail the backpressure promise
+        writeSecondPartPromise.succeed(())
+
+        XCTAssertEqual(delegate.receivedHead?.status, .movedPermanently)
+        XCTAssertNoThrow(try bag.task.futureResult.wait())
+    }
 }
 
 class MockRequestExecutor: HTTPRequestExecutor {
@@ -302,11 +367,15 @@ class MockRequestExecutor: HTTPRequestExecutor {
         case endOfStream
     }
 
+    let pauseRequestBodyPartStreamAfterASingleWrite: Bool
+
     private(set) var requestBodyParts = CircularBuffer<RequestParts>()
     private(set) var isCancelled: Bool = false
     private(set) var signalledDemandForResponseBody: Bool = false
 
-    init() {}
+    init(pauseRequestBodyPartStreamAfterASingleWrite: Bool = false) {
+        self.pauseRequestBodyPartStreamAfterASingleWrite = pauseRequestBodyPartStreamAfterASingleWrite
+    }
 
     func nextBodyPart() -> RequestParts? {
         guard !self.requestBodyParts.isEmpty else { return nil }
@@ -321,7 +390,7 @@ class MockRequestExecutor: HTTPRequestExecutor {
     // data is already scheduled. If we call pause here, once, after the second call new subsequent
     // calls should not be scheduled.
     func writeRequestBodyPart(_ part: IOData, request: HTTPExecutingRequest) {
-        if self.requestBodyParts.isEmpty {
+        if self.requestBodyParts.isEmpty, self.pauseRequestBodyPartStreamAfterASingleWrite {
             request.pauseRequestBodyStream()
         }
         self.requestBodyParts.append(.body(part))