Code review

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTPRequestStateMachine+Demand.swift

@@ -50,7 +50,7 @@ extension HTTPRequestStateMachine {
                 self.state = .modifying
                 buffer.append(body)
                 self.state = .waitingForBytes(buffer)
-                
+
             case .modifying:
                 preconditionFailure("Invalid state: \(self.state)")
 

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTPRequestStateMachine.swift

@@ -523,19 +523,31 @@ struct HTTPRequestStateMachine {
             where head.status.code < 300:
 
             return self.avoidingStateMachineCoW { state -> Action in
-                let (remainingBuffer, _) = responseStreamState.end()
-                state = .running(
-                    .streaming(expectedBodyLength: expectedBodyLength, sentBodyBytes: sentBodyBytes, producer: producerState),
-                    .endReceived
-                )
-                return .forwardResponseBodyParts(remainingBuffer)
+                let (remainingBuffer, connectionAction) = responseStreamState.end()
+                switch connectionAction {
+                case .none:
+                    state = .running(
+                        .streaming(expectedBodyLength: expectedBodyLength, sentBodyBytes: sentBodyBytes, producer: producerState),
+                        .endReceived
+                    )
+                    return .forwardResponseBodyParts(remainingBuffer)
+                case .close:
+                    // If we receive a `.close` as a connectionAction from the responseStreamState
+                    // this means, that the response end was signaled by a connection close. Since
+                    // the request is still uploading, we will not be able to finish the upload. For
+                    // this reason we can fail the request here.
+                    state = .failed(HTTPClientError.remoteConnectionClosed)
+                    return .failRequest(HTTPClientError.remoteConnectionClosed, .close)
+                }
             }
 
         case .running(.streaming(_, _, let producerState), .receivingBody(let head, var responseStreamState)):
             assert(head.status.code >= 300)
             assert(producerState == .paused, "Expected to have paused the request body stream, when the head was received. Invalid state: \(self.state)")
 
             return self.avoidingStateMachineCoW { state -> Action in
+                // We can ignore the connectionAction from the responseStreamState, since the
+                // connection should be closed anyway.
                 let (remainingBuffer, _) = responseStreamState.end()
                 state = .finished
                 return .succeedRequest(.close, remainingBuffer)

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

@@ -52,6 +52,7 @@ extension HTTPRequestStateMachineTests {
             ("testErrorWhileRunningARequestClosesTheStream", testErrorWhileRunningARequestClosesTheStream),
             ("testCanReadHTTP1_0ResponseWithoutBody", testCanReadHTTP1_0ResponseWithoutBody),
             ("testCanReadHTTP1_0ResponseWithBody", testCanReadHTTP1_0ResponseWithBody),
+            ("testFailHTTP1_0RequestThatIsStillUploading", testFailHTTP1_0RequestThatIsStillUploading),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTPRequestStateMachineTests.swift

@@ -485,6 +485,26 @@ class HTTPRequestStateMachineTests: XCTestCase {
         XCTAssertEqual(state.channelRead(.end(nil)), .succeedRequest(.close, [body]))
         XCTAssertEqual(state.channelInactive(), .wait)
     }
+
+    func testFailHTTP1_0RequestThatIsStillUploading() {
+        var state = HTTPRequestStateMachine(isChannelWritable: true)
+        let requestHead = HTTPRequestHead(version: .http1_1, method: .POST, uri: "/")
+        let metadata = RequestFramingMetadata(connectionClose: false, body: .stream)
+        XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: true))
+
+        let part1: ByteBuffer = .init(string: "foo")
+        XCTAssertEqual(state.requestStreamPartReceived(.byteBuffer(part1)), .sendBodyPart(.byteBuffer(part1)))
+        let responseHead = HTTPResponseHead(version: .http1_0, status: .ok)
+        let body = ByteBuffer(string: "foo bar")
+        XCTAssertEqual(state.channelRead(.head(responseHead)), .forwardResponseHead(responseHead, pauseRequestBodyStream: false))
+        XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
+        XCTAssertEqual(state.channelReadComplete(), .wait)
+        XCTAssertEqual(state.read(), .read)
+        XCTAssertEqual(state.channelReadComplete(), .wait)
+        XCTAssertEqual(state.channelRead(.body(body)), .wait)
+        XCTAssertEqual(state.channelRead(.end(nil)), .failRequest(HTTPClientError.remoteConnectionClosed, .close))
+        XCTAssertEqual(state.channelInactive(), .wait)
+    }
 }
 
 extension HTTPRequestStateMachine.Action: Equatable {