fail if user tries writing bytes after request is sent

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -967,6 +967,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
         case traceRequestWithBody
         case invalidHeaderFieldNames([String])
         case bodyLengthMismatch
+        case writeAfterRequestSent
     }
 
     private var code: Code
@@ -1019,4 +1020,6 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     public static func invalidHeaderFieldNames(_ names: [String]) -> HTTPClientError { return HTTPClientError(code: .invalidHeaderFieldNames(names)) }
     /// Body length is not equal to `Content-Length`.
     public static let bodyLengthMismatch = HTTPClientError(code: .bodyLengthMismatch)
+    /// Body part was written after request was fully sent.
+    public static let writeAfterRequestSent = HTTPClientError(code: .writeAfterRequestSent)
 }

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -876,12 +876,10 @@ extension TaskHandler: ChannelDuplexHandler {
                 let promise = self.task.eventLoop.makePromise(of: Void.self)
                 // All writes have to be switched to the channel EL if channel and task ELs differ
                 if channel.eventLoop.inEventLoop {
-                    self.actualBodyLength += part.readableBytes
-                    context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
+                    self.writeBodyPart(context: context, part: part, promise: promise)
                 } else {
                     channel.eventLoop.execute {
-                        self.actualBodyLength += part.readableBytes
-                        context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
+                        self.writeBodyPart(context: context, part: part, promise: promise)
                     }
                 }
 
@@ -901,6 +899,19 @@ extension TaskHandler: ChannelDuplexHandler {
         }
     }
 
+    private func writeBodyPart(context: ChannelHandlerContext, part: IOData, promise: EventLoopPromise<Void>) {
+        switch self.state {
+        case .idle:
+            self.actualBodyLength += part.readableBytes
+            context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
+        default:
+            let error = HTTPClientError.writeAfterRequestSent
+            promise.fail(error)
+            self.state = .endOrError
+            self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
+        }
+    }
+
     public func read(context: ChannelHandlerContext) {
         if self.mayRead {
             self.pendingRead = false

Tests/AsyncHTTPClientTests/HTTPClientInternalTests.swift

@@ -424,11 +424,10 @@ class HTTPClientInternalTests: XCTestCase {
         let randoEL = group.next()
 
         let httpClient = HTTPClient(eventLoopGroupProvider: .shared(group))
-        let promise: EventLoopPromise<Channel> = httpClient.eventLoopGroup.next().makePromise()
-        let httpBin = HTTPBin(channelPromise: promise)
+        let server = NIOHTTP1TestServer(group: MultiThreadedEventLoopGroup(numberOfThreads: 1))
         defer {
+            XCTAssertNoThrow(try server.stop())
             XCTAssertNoThrow(try httpClient.syncShutdown(requiresCleanClose: true))
-            XCTAssertNoThrow(try httpBin.shutdown())
         }
 
         let body: HTTPClient.Body = .stream(length: 8) { writer in
@@ -439,21 +438,26 @@ class HTTPClientInternalTests: XCTestCase {
             }
         }
 
-        let request = try Request(url: "http://127.0.0.1:\(httpBin.port)/custom",
+        let request = try Request(url: "http://127.0.0.1:\(server.serverPort)/custom",
                                   body: body)
         let delegate = Delegate(expectedEventLoop: delegateEL, randomOtherEventLoop: randoEL)
         let future = httpClient.execute(request: request,
                                         delegate: delegate,
                                         eventLoop: .init(.testOnly_exact(channelOn: channelEL,
                                                                          delegateOn: delegateEL))).futureResult
 
-        let channel = try promise.futureResult.wait()
+        XCTAssertNoThrow(try server.readInbound()) // .head
+        XCTAssertNoThrow(try server.readInbound()) // .body
+        XCTAssertNoThrow(try server.readInbound()) // .end
 
         // Send 3 parts, but only one should be received until the future is complete
-        let buffer = channel.allocator.buffer(string: "1234")
-        try channel.writeAndFlush(HTTPServerResponsePart.body(.byteBuffer(buffer))).wait()
+        XCTAssertNoThrow(try server.writeOutbound(.head(.init(version: .init(major: 1, minor: 1),
+                                                              status: .ok,
+                                                              headers: HTTPHeaders([("Transfer-Encoding", "chunked")])))))
+        let buffer = ByteBuffer(string: "1234")
+        XCTAssertNoThrow(try server.writeOutbound(.body(.byteBuffer(buffer))))
+        XCTAssertNoThrow(try server.writeOutbound(.end(nil)))
 
-        try channel.writeAndFlush(HTTPServerResponsePart.end(nil)).wait()
         let (receivedMessages, sentMessages) = try future.wait()
         XCTAssertEqual(2, receivedMessages.count)
         XCTAssertEqual(4, sentMessages.count)
@@ -488,7 +492,7 @@ class HTTPClientInternalTests: XCTestCase {
 
         switch receivedMessages.dropFirst(0).first {
         case .some(.head(let head)):
-            XCTAssertEqual(["transfer-encoding": "chunked"], head.headers)
+            XCTAssertEqual(head.headers["transfer-encoding"].first, "chunked")
         default:
             XCTFail("wrong message")
         }

Tests/AsyncHTTPClientTests/HTTPClientTests+XCTest.swift

@@ -118,6 +118,7 @@ extension HTTPClientTests {
             ("testDelegateCallinsTolerateRandomEL", testDelegateCallinsTolerateRandomEL),
             ("testContentLengthTooLongFails", testContentLengthTooLongFails),
             ("testContentLengthTooShortFails", testContentLengthTooShortFails),
+            ("testBodyUploadAfterEndFails", testBodyUploadAfterEndFails),
         ]
     }
 }

Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -2502,4 +2502,39 @@ class HTTPClientTests: XCTestCase {
         XCTAssertEqual(info.connectionNumber, 1)
         XCTAssertEqual(info.requestNumber, 1)
     }
+
+    func testBodyUploadAfterEndFails() {
+        let url = self.defaultHTTPBinURLPrefix + "post"
+
+        func uploader(_ streamWriter: HTTPClient.Body.StreamWriter) -> EventLoopFuture<Void> {
+            let done = streamWriter.write(.byteBuffer(ByteBuffer(string: "X")))
+            done.recover { error -> Void in
+                XCTFail("unexpected error \(error)")
+            }.whenSuccess {
+                // This is executed when we have already sent the end of the request.
+                done.eventLoop.execute {
+                    streamWriter.write(.byteBuffer(ByteBuffer(string: "BAD BAD BAD"))).whenComplete { result in
+                        switch result {
+                        case .success:
+                            XCTFail("we succeeded writing bytes after the end!?")
+                        case .failure(let error):
+                            XCTAssertEqual(HTTPClientError.writeAfterRequestSent, error as? HTTPClientError)
+                        }
+                    }
+                }
+            }
+            return done
+        }
+
+        XCTAssertThrowsError(
+            try self.defaultClient.execute(request:
+                Request(url: url,
+                        body: .stream(length: 1, uploader))).wait()) { error in
+            XCTAssertEqual(HTTPClientError.writeAfterRequestSent, error as? HTTPClientError)
+        }
+
+        // Quickly try another request and check that it works. If we by accident wrote some extra bytes into the
+        // stream (and reuse the connection) that could cause problems.
+        XCTAssertNoThrow(try self.defaultClient.get(url: self.defaultHTTPBinURLPrefix + "get").wait())
+    }
 }