Call didSendRequestPart at the right time

# Motivation
Right now, we call `didSendRequestPart` after passing the write to the executor. However, this does not mean that the write hit the socket. To implement proper backpressure using the delegate, we should only call this method once the write was successful.

# Modification
Pass a promise to the actual channel write and only call the delegate once that promise succeeds.

# Result
The delegate method `didSendRequestPart` is only called after the write was successful.

Author: FranzBusch

Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift

@@ -63,7 +63,7 @@ final class Transaction: @unchecked Sendable {
 
         switch writeAction {
         case .writeAndWait(let executor), .writeAndContinue(let executor):
-            executor.writeRequestBodyPart(.byteBuffer(byteBuffer), request: self)
+            executor.writeRequestBodyPart(.byteBuffer(byteBuffer), request: self, promise: nil)
 
         case .fail:
             // an error/cancellation has happened. we don't need to continue here
@@ -105,14 +105,14 @@ final class Transaction: @unchecked Sendable {
         switch self.state.writeNextRequestPart() {
         case .writeAndContinue(let executor):
             self.stateLock.unlock()
-            executor.writeRequestBodyPart(.byteBuffer(part), request: self)
+            executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil)
 
         case .writeAndWait(let executor):
             try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
                 self.state.waitForRequestBodyDemand(continuation: continuation)
                 self.stateLock.unlock()
 
-                executor.writeRequestBodyPart(.byteBuffer(part), request: self)
+                executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil)
             }
 
         case .fail:
@@ -132,7 +132,7 @@ final class Transaction: @unchecked Sendable {
             break
 
         case .forwardStreamFinished(let executor, let succeedContinuation):
-            executor.finishRequestBodyStream(self)
+            executor.finishRequestBodyStream(self, promise: nil)
             succeedContinuation?.resume(returning: nil)
         }
         return

Sources/AsyncHTTPClient/ConnectionPool/HTTP1/HTTP1ClientChannelHandler.swift

@@ -185,11 +185,11 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
         case .sendRequestHead(let head, startBody: let startBody):
             self.sendRequestHead(head, startBody: startBody, context: context)
 
-        case .sendBodyPart(let part):
-            context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: nil)
+        case .sendBodyPart(let part, let writePromise):
+            context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: writePromise)
 
-        case .sendRequestEnd:
-            context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
+        case .sendRequestEnd(let writePromise):
+            context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: writePromise)
 
             if let timeoutAction = self.idleReadTimeoutStateMachine?.requestEndSent() {
                 self.runTimeoutAction(timeoutAction, context: context)
@@ -258,36 +258,58 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
             self.runTimeoutAction(.clearIdleReadTimeoutTimer, context: context)
 
             switch finalAction {
-            case .close:
+            case .close(let writePromise):
                 context.close(promise: nil)
-            case .sendRequestEnd:
-                context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
-            case .informConnectionIsIdle:
+                writePromise?.succeed(())
+                oldRequest.succeedRequest(buffer)
+            case .sendRequestEnd(let writePromise):
+                let writePromise = writePromise ?? context.eventLoop.makePromise(of: Void.self)
+                // We need to defer succeeding the old request to avoid ordering issues
+                writePromise.futureResult.whenComplete { _ in
+                    oldRequest.succeedRequest(buffer)
+                }
+
+                context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: writePromise)
+            case .informConnectionIsIdle(let writePromise):
                 self.connection.taskCompleted()
+                writePromise?.succeed(())
+                oldRequest.succeedRequest(buffer)
             case .none:
-                break
+                oldRequest.succeedRequest(buffer)
             }
 
-            oldRequest.succeedRequest(buffer)
-
         case .failRequest(let error, let finalAction):
             // see comment in the `succeedRequest` case.
             let oldRequest = self.request!
             self.request = nil
             self.runTimeoutAction(.clearIdleReadTimeoutTimer, context: context)
 
             switch finalAction {
-            case .close:
-                context.close(promise: nil)
-            case .sendRequestEnd:
-                context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
-            case .informConnectionIsIdle:
+            case .close(let writePromise):
+                context.close(promise: writePromise)
+                writePromise?.fail(error)
+                oldRequest.fail(error)
+
+            case .sendRequestEnd(let writePromise):
+                let writePromise = writePromise ?? context.eventLoop.makePromise(of: Void.self)
+                // We need to defer failing the old request to avoid ordering issues
+                writePromise.futureResult.whenComplete { _ in
+                    oldRequest.fail(error)
+                }
+
+                context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: writePromise)
+
+            case .informConnectionIsIdle(let writePromise):
                 self.connection.taskCompleted()
+                writePromise?.fail(error)
+                oldRequest.fail(error)
+
             case .none:
-                break
+                oldRequest.fail(error)
             }
 
-            oldRequest.fail(error)
+        case .failSendBodyPart(let error, let writePromise), .failSendStreamFinished(let error, let writePromise):
+            writePromise?.fail(error)
         }
     }
 
@@ -355,27 +377,28 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
 
     // MARK: Private HTTPRequestExecutor
 
-    private func writeRequestBodyPart0(_ data: IOData, request: HTTPExecutableRequest) {
+    private func writeRequestBodyPart0(_ data: IOData, request: HTTPExecutableRequest, promise: EventLoopPromise<Void>?) {
         guard self.request === request, let context = self.channelContext else {
             // Because the HTTPExecutableRequest may run in a different thread to our eventLoop,
             // calls from the HTTPExecutableRequest to our ChannelHandler may arrive here after
             // the request has been popped by the state machine or the ChannelHandler has been
             // removed from the Channel pipeline. This is a normal threading issue, noone has
             // screwed up.
+            promise?.fail(HTTPClientError.requestStreamCancelled)
             return
         }
 
-        let action = self.state.requestStreamPartReceived(data)
+        let action = self.state.requestStreamPartReceived(data, promise: promise)
         self.run(action, context: context)
     }
 
-    private func finishRequestBodyStream0(_ request: HTTPExecutableRequest) {
+    private func finishRequestBodyStream0(_ request: HTTPExecutableRequest, promise: EventLoopPromise<Void>?) {
         guard self.request === request, let context = self.channelContext else {
             // See code comment in `writeRequestBodyPart0`
             return
         }
 
-        let action = self.state.requestStreamFinished()
+        let action = self.state.requestStreamFinished(promise: promise)
         self.run(action, context: context)
     }
 
@@ -405,22 +428,22 @@ final class HTTP1ClientChannelHandler: ChannelDuplexHandler {
 }
 
 extension HTTP1ClientChannelHandler: HTTPRequestExecutor {
-    func writeRequestBodyPart(_ data: IOData, request: HTTPExecutableRequest) {
+    func writeRequestBodyPart(_ data: IOData, request: HTTPExecutableRequest, promise: EventLoopPromise<Void>?) {
         if self.eventLoop.inEventLoop {
-            self.writeRequestBodyPart0(data, request: request)
+            self.writeRequestBodyPart0(data, request: request, promise: promise)
         } else {
             self.eventLoop.execute {
-                self.writeRequestBodyPart0(data, request: request)
+                self.writeRequestBodyPart0(data, request: request, promise: promise)
             }
         }
     }
 
-    func finishRequestBodyStream(_ request: HTTPExecutableRequest) {
+    func finishRequestBodyStream(_ request: HTTPExecutableRequest, promise: EventLoopPromise<Void>?) {
         if self.eventLoop.inEventLoop {
-            self.finishRequestBodyStream0(request)
+            self.finishRequestBodyStream0(request, promise: promise)
         } else {
             self.eventLoop.execute {
-                self.finishRequestBodyStream0(request)
+                self.finishRequestBodyStream0(request, promise: promise)
             }
         }
     }

Sources/AsyncHTTPClient/ConnectionPool/HTTP1/HTTP1ConnectionStateMachine.swift

@@ -30,19 +30,29 @@ struct HTTP1ConnectionStateMachine {
         /// A action to execute, when we consider a request "done".
         enum FinalStreamAction {
             /// Close the connection
-            case close
+            ///
+            /// The promise is an optional write promise.
+            case close(EventLoopPromise<Void>?)
             /// If the server has replied, with a status of 200...300 before all data was sent, a request is considered succeeded,
             /// as soon as we wrote the request end onto the wire.
-            case sendRequestEnd
+            ///
+            /// The promise is an optional write promise.
+            case sendRequestEnd(EventLoopPromise<Void>?)
             /// Inform an observer that the connection has become idle
-            case informConnectionIsIdle
+            ///
+            /// The promise is an optional write promise.
+            case informConnectionIsIdle(EventLoopPromise<Void>?)
             /// Do nothing.
-            case none
+            ///
+            /// The promise is an optional write promise.
+            case none(EventLoopPromise<Void>?)
         }
 
         case sendRequestHead(HTTPRequestHead, startBody: Bool)
-        case sendBodyPart(IOData)
-        case sendRequestEnd
+        case sendBodyPart(IOData, EventLoopPromise<Void>?)
+        case sendRequestEnd(EventLoopPromise<Void>?)
+        case failSendBodyPart(Error, EventLoopPromise<Void>?)
+        case failSendStreamFinished(Error, EventLoopPromise<Void>?)
 
         case pauseRequestBodyStream
         case resumeRequestBodyStream
@@ -173,7 +183,7 @@ struct HTTP1ConnectionStateMachine {
             // as closed.
             //
             // TODO: AHC should support a fast rescheduling mechanism here.
-            return .failRequest(HTTPClientError.remoteConnectionClosed, .none)
+            return .failRequest(HTTPClientError.remoteConnectionClosed, .none(nil))
 
         case .idle:
             var requestStateMachine = HTTPRequestStateMachine(isChannelWritable: self.isChannelWritable)
@@ -189,25 +199,25 @@ struct HTTP1ConnectionStateMachine {
         }
     }
 
-    mutating func requestStreamPartReceived(_ part: IOData) -> Action {
+    mutating func requestStreamPartReceived(_ part: IOData, promise: EventLoopPromise<Void>?) -> Action {
         guard case .inRequest(var requestStateMachine, let close) = self.state else {
             preconditionFailure("Invalid state: \(self.state)")
         }
 
         return self.avoidingStateMachineCoW { state -> Action in
-            let action = requestStateMachine.requestStreamPartReceived(part)
+            let action = requestStateMachine.requestStreamPartReceived(part, promise: promise)
             state = .inRequest(requestStateMachine, close: close)
             return state.modify(with: action)
         }
     }
 
-    mutating func requestStreamFinished() -> Action {
+    mutating func requestStreamFinished(promise: EventLoopPromise<Void>?) -> Action {
         guard case .inRequest(var requestStateMachine, let close) = self.state else {
             preconditionFailure("Invalid state: \(self.state)")
         }
 
         return self.avoidingStateMachineCoW { state -> Action in
-            let action = requestStateMachine.requestStreamFinished()
+            let action = requestStateMachine.requestStreamFinished(promise: promise)
             state = .inRequest(requestStateMachine, close: close)
             return state.modify(with: action)
         }
@@ -377,10 +387,10 @@ extension HTTP1ConnectionStateMachine.State {
             return .pauseRequestBodyStream
         case .resumeRequestBodyStream:
             return .resumeRequestBodyStream
-        case .sendBodyPart(let part):
-            return .sendBodyPart(part)
-        case .sendRequestEnd:
-            return .sendRequestEnd
+        case .sendBodyPart(let part, let writePromise):
+            return .sendBodyPart(part, writePromise)
+        case .sendRequestEnd(let writePromise):
+            return .sendRequestEnd(writePromise)
         case .forwardResponseHead(let head, let pauseRequestBodyStream):
             return .forwardResponseHead(head, pauseRequestBodyStream: pauseRequestBodyStream)
         case .forwardResponseBodyParts(let parts):
@@ -392,39 +402,63 @@ extension HTTP1ConnectionStateMachine.State {
 
             let newFinalAction: HTTP1ConnectionStateMachine.Action.FinalStreamAction
             switch finalAction {
-            case .close:
+            case .close(let writePromise):
                 self = .closing
-                newFinalAction = .close
-            case .sendRequestEnd:
-                newFinalAction = .sendRequestEnd
+                newFinalAction = .close(writePromise)
+            case .sendRequestEnd(let writePromise):
+                newFinalAction = .sendRequestEnd(writePromise)
             case .none:
                 self = .idle
-                newFinalAction = close ? .close : .informConnectionIsIdle
+                newFinalAction = close ? .close(nil) : .informConnectionIsIdle(nil)
             }
             return .succeedRequest(newFinalAction, finalParts)
 
         case .failRequest(let error, let finalAction):
-            switch self {
-            case .initialized:
+            switch (self, finalAction) {
+            case (.initialized, _):
                 preconditionFailure("Invalid state: \(self)")
-            case .idle:
+
+            case (.idle, _):
                 preconditionFailure("How can we fail a task, if we are idle")
-            case .inRequest(_, close: let close):
-                if close || finalAction == .close {
-                    self = .closing
-                    return .failRequest(error, .close)
-                } else {
-                    self = .idle
-                    return .failRequest(error, .informConnectionIsIdle)
-                }
 
-            case .closing:
-                return .failRequest(error, .none)
-            case .closed:
+            // If we are either in .inRequest(_, close: true) or the final action is .close
+            // we have to fail the request with .close()
+            case (.inRequest(_, let close), .sendRequestEnd(let writePromise)) where close:
+                self = .closing
+                return .failRequest(error, .close(writePromise))
+
+            case (.inRequest(_, let close), .none) where close:
+                self = .closing
+                return .failRequest(error, .close(nil))
+
+            case (.inRequest(_, _), .close(let writePromise)):
+                self = .closing
+                return .failRequest(error, .close(writePromise))
+
+            // otherwise we fail with .informConnectionIsIdle
+            case (.inRequest(_, _), .sendRequestEnd(let writePromise)):
+                self = .idle
+                return .failRequest(error, .informConnectionIsIdle(writePromise))
+
+            case (.inRequest(_, _), .none):
+                self = .idle
+                return .failRequest(error, .informConnectionIsIdle(nil))
+
+            case (.closing, .close(let writePromise)), (.closing, .sendRequestEnd(let writePromise)):
+                return .failRequest(error, .none(writePromise))
+
+            case (.closing, .none):
+                return .failRequest(error, .none(nil))
+
+            case (.closed, .close(let writePromise)), (.closed, .sendRequestEnd(let writePromise)):
                 // this state can be reached, if the connection was unexpectedly closed by remote
-                return .failRequest(error, .none)
+                return .failRequest(error, .none(writePromise))
 
-            case .modifying:
+            case (.closed, .none):
+                // this state can be reached, if the connection was unexpectedly closed by remote
+                return .failRequest(error, .none(nil))
+
+            case (.modifying, _):
                 preconditionFailure("Invalid state: \(self)")
             }
 
@@ -433,6 +467,12 @@ extension HTTP1ConnectionStateMachine.State {
 
         case .wait:
             return .wait
+
+        case .failSendBodyPart(let error, let writePromise):
+            return .failSendBodyPart(error, writePromise)
+
+        case .failSendStreamFinished(let error, let writePromise):
+            return .failSendStreamFinished(error, writePromise)
         }
     }
 }