Move handling of inputClosed to HTTP3StreamHandler

Author: aryan-25

Sources/HTTP3/HTTP3Error.swift

@@ -140,7 +140,7 @@ extension HTTP3Error {
             case none
             case invalidGoawayStreamID
             case criticalStreamClosed
-            case peerTerminatedStream
+            case peerTerminatedInboundStream
         }
 
         public var description: String {
@@ -252,9 +252,9 @@ extension HTTP3Error {
             Self(.criticalStreamClosed)
         }
 
-        /// The peer terminated the stream before delivering a complete request or response.
-        public static var peerTerminatedStream: Self {
-            Self(.peerTerminatedStream)
+        /// The peer terminated the inbound stream before delivering a complete request or response.
+        public static var peerTerminatedInboundStream: Self {
+            Self(.peerTerminatedInboundStream)
         }
     }
 

Sources/HTTP3/HTTP3FrameValidator.swift

@@ -176,7 +176,7 @@ package enum HTTP3FrameValidator: ~Copyable {
     /// It looks at both request and response frames.
     /// When using this validator for a client, the request frames are the outbound and the response frames are inbound.
     /// This is reversed for a server.
-    package struct RequestStreamValidator: ~Copyable {
+    private struct RequestStreamValidator: ~Copyable {
         /// Models the state of one side of the connection.
         enum State {
             /// Nothing has happened yet.
@@ -357,30 +357,151 @@ package enum HTTP3FrameValidator: ~Copyable {
                 }
             }
         }
+
+        /// Whether the complete request has been received.
+        private var receivedCompleteRequest: Bool {
+            switch self.requestState {
+            case .idle:
+                // We haven't received the request head part yet, so the request is trivially incomplete.
+                return false
+
+            case .headersProcessed:
+                // TODO: If a Content-Length header was specified, we need to check whether we have received all the
+                // specified bytes. If that is not the case, then the request is malformed per RFC 9114 § 4.1.2, and it
+                // must be treated as a stream error.
+                return true
+
+            case .trailersProcessed:
+                // If we have processed trailers, we have seen the full request.
+                return true
+
+            case .previousError:
+                // There was already an error. That will result in the stream closing anyway, so just return `true`.
+                return true
+            }
+        }
+
+        /// Whether the complete response has been received.
+        private var receivedCompleteResponse: Bool {
+            switch self.responseState {
+            case .idle:
+                // We haven't received a final response head part yet, so the response is trivially incomplete.
+                return false
+
+            case .headersProcessed:
+                // TODO: If a Content-Length header was specified, we need to check whether we have received all the
+                // specified bytes. If that is not the case, then the request is malformed per RFC 9114 § 4.1.2, and it
+                // must be treated as a stream error.
+                return true
+
+            case .trailersProcessed:
+                // If we have processed trailers, we have seen the full response.
+                return true
+
+            case .previousError:
+                // There was already an error. That will result in the stream closing anyway, so just return `true`.
+                return true
+            }
+        }
+
+        /// Records that the inbound request stream has closed.
+        mutating func processInboundRequestStreamClosed() -> InboundClosedAction {
+            if self.receivedCompleteRequest {
+                return .doNothing
+            }
+
+            // The input closed before we saw a *complete* request. Per RFC 9114 § 4.1:
+            // "If a client-initiated stream terminates without enough of the HTTP message to provide a complete
+            //  response, the server SHOULD abort its response stream with the error code H3\_REQUEST\_INCOMPLETE".
+            //
+            // We therefore need to reset the stream.
+            self = .init(requestState: .previousError, responseState: .previousError)
+
+            return .resetStream(
+                HTTP3Error(
+                    code: .peerTerminatedInboundStream,
+                    message: "Inbound request stream closed before a complete request was received",
+                    cause: nil,
+                    errorCode: .H3_REQUEST_INCOMPLETE,
+                    location: .here()
+                )
+            )
+        }
+
+        /// Records that the inbound response stream has closed.
+        func processInboundResponseStreamClosed() -> InboundClosedAction {
+            if self.receivedCompleteResponse {
+                return .doNothing
+            }
+
+            // The input closed before we saw a *complete* response. Per RFC 9114 § 4.1.1:
+            // "... if a stream is cancelled after receiving a partial response, the response SHOULD NOT be used".
+            //
+            // We therefore need to inform the downstream so they can decide what to do with the incomplete response.
+            //
+            // This is not a stream or connection error, so we don't modify the request or response state.
+            return .notifyDownstream(
+                HTTP3Error(
+                    code: .peerTerminatedInboundStream,
+                    message: "Inbound response stream closed before a complete response was received",
+                    cause: nil,
+                    errorCode: .H3_NO_ERROR,
+                    location: .here()
+                )
+            )
+        }
+    }
+
+    /// The action to take when the inbound side of a stream closes.
+    package enum InboundClosedAction {
+        /// We received the full request or response before the inbound closed. As such, there is nothing to do.
+        case doNothing
+
+        /// The inbound closed before we received the full response. The downstream should be notified so they can
+        /// decide what to do with the partial response.
+        case notifyDownstream(HTTP3Error)
+
+        /// The inbound closed before we received the full request. Per RFC 9114 § 4.1, the server should abort the
+        /// response stream by sending a RESET_STREAM frame.
+        case resetStream(HTTP3Error)
     }
 
     package struct ServerRequestStreamValidator: ~Copyable {
         private var underlying = RequestStreamValidator()
 
+        /// Validates an inbound request frame.
         package mutating func processInboundFrame(_ frame: HTTP3Frame) -> ProcessFrameAction {
             self.underlying.processRequestFrame(frame)
         }
 
+        /// Validates an outbound response frame.
         package mutating func processOutboundFrame(_ frame: HTTP3Frame) -> ProcessFrameAction {
             self.underlying.processResponseFrame(frame)
         }
+
+        /// Records that the inbound request stream has closed.
+        package mutating func processInboundClosed() -> InboundClosedAction {
+            self.underlying.processInboundRequestStreamClosed()
+        }
     }
 
     package struct ClientRequestStreamValidator: ~Copyable {
         private var underlying = RequestStreamValidator()
 
+        /// Validates an inbound response frame.
         package mutating func processInboundFrame(_ frame: HTTP3Frame) -> ProcessFrameAction {
             self.underlying.processResponseFrame(frame)
         }
 
+        /// Validates an outbound request frame.
         package mutating func processOutboundFrame(_ frame: HTTP3Frame) -> ProcessFrameAction {
             self.underlying.processRequestFrame(frame)
         }
+
+        /// Records that the inbound response stream has closed.
+        package func processInboundClosed() -> InboundClosedAction {
+            self.underlying.processInboundResponseStreamClosed()
+        }
     }
 
     /// Checks that incoming frames are in the right order for a control stream, and that there are no outgoing frames.
@@ -594,6 +715,37 @@ package enum HTTP3FrameValidator: ~Copyable {
             return .dropFrame
         }
     }
+
+    /// Records that the inbound side of the stream has closed.
+    package mutating func processInboundClosed() -> InboundClosedAction {
+        switch self {
+        case .incomingRequestStream(var validator):
+            let result = validator.processInboundClosed()
+            self = .incomingRequestStream(validator)
+            return result
+
+        case .outgoingRequestStream(let validator):
+            let result = validator.processInboundClosed()
+            self = .outgoingRequestStream(validator)
+            return result
+
+        case .incomingControlStream(let validator):
+            self = .incomingControlStream(validator)
+            return .doNothing
+
+        case .outgoingControlStream(let validator):
+            self = .outgoingControlStream(validator)
+            return .doNothing
+
+        case .incomingPushStream(let validator):
+            self = .incomingPushStream(validator)
+            return .doNothing
+
+        case .outgoingPushStream(let validator):
+            self = .outgoingPushStream(validator)
+            return .doNothing
+        }
+    }
 }
 
 extension HTTP3Frame {

Sources/HTTP3/HTTP3StreamStateMachine.swift

@@ -195,7 +195,7 @@ package struct HTTP3StreamStateMachine: ~Copyable {
             }
         }
 
-        /// Inform the state machine of a qpack decode result that has been previously been asked for.
+        /// Inform the state machine of a qpack decode result that has been previously asked for.
         /// It is an error to call this function with a result for a partial header which wasn't asked for.
         mutating func gotHeaderDecodeResult(_ decoded: [HTTPField], from: HTTP3PartialFrame.Headers) {
             switch consume self.state {
@@ -577,7 +577,7 @@ package struct HTTP3StreamStateMachine: ~Copyable {
         /// A frame is ready, but you need to decode it and call the state machine back with the result.
         case decodeHeader(HTTP3PartialFrame.Headers)
         /// The input was newly closed.
-        case inputClosed
+        case inputClosed(InputClosedAction)
         /// The input was already closed
         case alreadyClosed
         /// More input is needed before the next action can be determined
@@ -586,6 +586,19 @@ package struct HTTP3StreamStateMachine: ~Copyable {
         case previousError
         /// The decodeNext() function should be called again to get the next action.
         case callAgain
+
+        package enum InputClosedAction {
+            /// A complete request/response was received before the input was closed. As such, we should just deliver
+            /// the `inputClosed` event downstream.
+            case emitEvent
+
+            /// A complete response was not received before the input was closed. We need to notify the downstream about
+            /// the incompleteness through an error and then deliver the `inputClosed` event.
+            case emitErrorAndEvent(HTTP3Error)
+
+            /// A complete request was not received before the input was closed. We need to send a RESET\_STREAM frame.
+            case resetStream(HTTP3Error)
+        }
     }
 
     /// Read out the next frame if it is ready. This may ask you to run qpack on some partial headers.
@@ -643,8 +656,20 @@ package struct HTTP3StreamStateMachine: ~Copyable {
                 self = .init(state: .idle(idleState))
                 return .needMoreBytes
             case .inputClosed:
-                self = .init(state: .idle(idleState))
-                return .inputClosed
+                // The input was closed. Inform the validator to determine what to do next.
+                switch idleState.validator.processInboundClosed() {
+                case .doNothing:
+                    self = .init(state: .idle(idleState))
+                    return .inputClosed(.emitEvent)
+
+                case .notifyDownstream(let error):
+                    self = .init(state: .idle(idleState))
+                    return .inputClosed(.emitErrorAndEvent(error))
+
+                case .resetStream(let error):
+                    self = .init(state: .idle(idleState))
+                    return .inputClosed(.resetStream(error))
+                }
             }
         case .finished:
             self = .init(state: .finished)

Sources/NIOHTTP3/HTTP3StreamHandler.swift

@@ -116,8 +116,8 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler {
             var didFireChannelRead = false
             loop: for action in actionBuffer {
                 switch action {
-                case .inputClosed:
-                    context.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+                case .inputClosed(let inputClosedAction):
+                    self.handleInputClosed(action: inputClosedAction, context: context)
                 case .returnFrame(let frame):
                     context.fireChannelRead(wrapInboundOut(frame))
                     didFireChannelRead = true
@@ -167,8 +167,8 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler {
         decodeLoop: while true {
             let action = self.stateMachine.decodeNext()
             switch action {
-            case .inputClosed:
-                context.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+            case .inputClosed(let inputClosedAction):
+                self.handleInputClosed(action: inputClosedAction, context: context)
             case .needMoreBytes, .alreadyClosed, .previousError:
                 break decodeLoop
             case .callAgain:
@@ -282,6 +282,29 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler {
         }
     }
 
+    /// Processes the `InputClosedAction`.
+    private func handleInputClosed(
+        action: HTTP3StreamStateMachine.DecodeNextAction.InputClosedAction,
+        context: ChannelHandlerContext
+    ) {
+        switch action {
+        case .emitEvent:
+            context.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+
+        case .emitErrorAndEvent(let error):
+            context.fireErrorCaught(error)
+            context.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+
+        case .resetStream(let error):
+            context.triggerUserOutboundEvent(
+                QUICResetStreamEvent(code: QUICApplicationErrorCode(error.h3ErrorCode ?? .H3_NO_ERROR)),
+                promise: nil
+            )
+            context.fireErrorCaught(error)
+            context.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+        }
+    }
+
     package func errorCaught(context: ChannelHandlerContext, error: any Error) {
         switch error {
         case let error as QUICStreamResetError:
@@ -311,7 +334,7 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler {
     package func onQPACKDecodeResult(fields: [HTTPField], forHeaders headers: HTTP3PartialFrame.Headers) {
         self.logger.trace("HTTP3StreamHandler.onQPACKDecodeResult")
         guard let context = self.context else {
-            // The stream must have been created an registered to get QPACK events and thus already have
+            // The stream must have been created and registered to get QPACK events and thus already have
             // the context available. Since pending decodes are dropped when the stream closes it must
             // still be open and active.
             fatalError("Tried to deliver QPACK results before handler was added")