Failing tests

Author: fabianfett

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

@@ -51,10 +51,28 @@ extension HTTPRequestStateMachine {
                 buffer.append(body)
                 self.state = .waitingForBytes(buffer)
 
-            case .waitingForRead,
-                 .waitingForDemand,
-                 .waitingForReadOrDemand:
-                preconditionFailure("How can we receive a body part, after a channelReadComplete, but no read has been forwarded yet. Invalid state: \(self.state)")
+            // For all the following cases, please note:
+            // Normally these code paths should never be hit. However there is one way to trigger
+            // this:
+            //
+            // If the server decides to close a connection, NIO will forward all outstanding
+            // `channelRead`s without waiting for a next `context.read` call. For this reason we
+            // might receive further bytes, when we don't expect them here.
+
+            case .waitingForRead(var buffer):
+                self.state = .modifying
+                buffer.append(body)
+                self.state = .waitingForRead(buffer)
+
+            case .waitingForDemand(var buffer):
+                self.state = .modifying
+                buffer.append(body)
+                self.state = .waitingForDemand(buffer)
+
+            case .waitingForReadOrDemand(var buffer):
+                self.state = .modifying
+                buffer.append(body)
+                self.state = .waitingForReadOrDemand(buffer)
 
             case .modifying:
                 preconditionFailure("Invalid state: \(self.state)")
@@ -134,15 +152,26 @@ extension HTTPRequestStateMachine {
             }
         }
 
-        mutating func end() -> CircularBuffer<ByteBuffer> {
+        enum ConnectionAction {
+            case none
+            case close
+        }
+
+        mutating func end() -> (CircularBuffer<ByteBuffer>, ConnectionAction) {
             switch self.state {
             case .waitingForBytes(let buffer):
-                return buffer
-
-            case .waitingForReadOrDemand,
-                 .waitingForRead,
-                 .waitingForDemand:
-                preconditionFailure("How can we receive a body end, after a channelReadComplete, but no read has been forwarded yet. Invalid state: \(self.state)")
+                return (buffer, .none)
+
+            case .waitingForReadOrDemand(let buffer),
+                 .waitingForRead(let buffer),
+                 .waitingForDemand(let buffer):
+                // Normally this code path should never be hit. However there is one way to trigger
+                // this:
+                //
+                // If the server decides to close a connection, NIO will forward all outstanding
+                // `channelRead`s without waiting for a next `context.read` call. For this reason we
+                // might receive a call to `end()`, when we don't expect it here.
+                return (buffer, .close)
 
             case .modifying:
                 preconditionFailure("Invalid state: \(self.state)")

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

@@ -523,7 +523,7 @@ struct HTTPRequestStateMachine {
             where head.status.code < 300:
 
             return self.avoidingStateMachineCoW { state -> Action in
-                let remainingBuffer = responseStreamState.end()
+                let (remainingBuffer, _) = responseStreamState.end()
                 state = .running(
                     .streaming(expectedBodyLength: expectedBodyLength, sentBodyBytes: sentBodyBytes, producer: producerState),
                     .endReceived
@@ -536,16 +536,21 @@ struct HTTPRequestStateMachine {
             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
-                let remainingBuffer = responseStreamState.end()
+                let (remainingBuffer, _) = responseStreamState.end()
                 state = .finished
                 return .succeedRequest(.close, remainingBuffer)
             }
 
         case .running(.endSent, .receivingBody(_, var responseStreamState)):
             return self.avoidingStateMachineCoW { state -> Action in
-                let remainingBuffer = responseStreamState.end()
+                let (remainingBuffer, action) = responseStreamState.end()
                 state = .finished
-                return .succeedRequest(.none, remainingBuffer)
+                switch action {
+                case .none:
+                    return .succeedRequest(.none, remainingBuffer)
+                case .close:
+                    return .succeedRequest(.close, remainingBuffer)
+                }
             }
 
         case .running(_, .endReceived), .finished:

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

@@ -50,6 +50,8 @@ extension HTTPRequestStateMachineTests {
             ("testReadTimeoutThatFiresToLateIsIgnored", testReadTimeoutThatFiresToLateIsIgnored),
             ("testCancellationThatIsInvokedToLateIsIgnored", testCancellationThatIsInvokedToLateIsIgnored),
             ("testErrorWhileRunningARequestClosesTheStream", testErrorWhileRunningARequestClosesTheStream),
+            ("testCanReadHTTP1_0ResponseWithoutBody", testCanReadHTTP1_0ResponseWithoutBody),
+            ("testCanReadHTTP1_0ResponseWithBody", testCanReadHTTP1_0ResponseWithBody),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTPRequestStateMachineTests.swift

@@ -451,6 +451,40 @@ class HTTPRequestStateMachineTests: XCTestCase {
         XCTAssertEqual(state.errorHappened(NIOSSLError.uncleanShutdown), .failRequest(NIOSSLError.uncleanShutdown, .close))
         XCTAssertEqual(state.requestCancelled(), .wait, "A cancellation that happens to late is ignored")
     }
+
+    func testCanReadHTTP1_0ResponseWithoutBody() {
+        var state = HTTPRequestStateMachine(isChannelWritable: true)
+        let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
+        let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
+        XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
+
+        let responseHead = HTTPResponseHead(version: .http1_0, status: .internalServerError)
+        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(.end(nil)), .succeedRequest(.close, []))
+        XCTAssertEqual(state.channelInactive(), .wait)
+    }
+
+    func testCanReadHTTP1_0ResponseWithBody() {
+        var state = HTTPRequestStateMachine(isChannelWritable: true)
+        let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
+        let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
+        XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
+
+        let responseHead = HTTPResponseHead(version: .http1_0, status: .internalServerError)
+        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)), .succeedRequest(.close, [body]))
+        XCTAssertEqual(state.channelInactive(), .wait)
+    }
 }
 
 extension HTTPRequestStateMachine.Action: Equatable {