Correctly handle inputClosed event (#13)

### Motivation

In `HTTP3ToHTTPServerCodec` and `HTTP3ToHTTPClientCodec`, we currently
emit a response/request end part if the client/server terminated the
stream with an incomplete response/request. This means that downstream
handlers can receive a response/request end part before receiving a
head/body part.

Per [RFC 9114
§4.1](https://datatracker.ietf.org/doc/html/rfc9114#section-4.1-14),
servers should abort the response stream with the
`H3_INCOMPLETE_REQUEST` error code when a client-initiated stream is
terminated before receiving enough request parts to provide a full
response. Additionally,
[§4.1.1](https://datatracker.ietf.org/doc/html/rfc9114#section-4.1.1-6)
describes that clients should not use partial responses.

### Modifications

- When we receive the `.inputClosed` event, `HTTP3FrameValidator` now
checks whether it received a complete request/response, and returns one
of three actions: `.doNothing` (if we received a complete
request/response), `.notifyDownstream` (if we received an incomplete
response (clients only)), or `.resetStream` (if we received an
incomplete request (servers only)).
- `HTTP3StreamStateMachine` now passes that action as an associated
value on the `inputClosed` action.
- `HTTP3StreamHandler` fires the `inputClosed` event as before, but for
an incomplete message it first fires an `HTTP3Error` down the pipeline,
and for servers, it also sends a `QUICResetStreamEvent` with the
`H3_REQUEST_INCOMPLETE` code.
- `HTTPMessageParsingStateMachine` no longer emits an end part when the
input closes before a head part arrives.
- Added a `peerTerminatedInboundStream` case to `HTTP3Error.Code`.

### Result

`HTTP3ToHTTPServerCodec` and `HTTP3ToHTTPClientCodec` no longer emit a
request/response end part if the stream has terminated without a
complete request/response.

Instead, `HTTP3StreamHandler` handler notifies downstream handlers with
an error, and for servers, the stream is also reset.

Author: aryan-25

Sources/HTTP3/HTTP3Error.swift

@@ -144,6 +144,7 @@ extension HTTP3Error {
             case none
             case invalidGoawayStreamID
             case criticalStreamClosed
+            case peerTerminatedInboundStream
         }
 
         public var description: String {
@@ -254,6 +255,11 @@ extension HTTP3Error {
         public static var criticalStreamClosed: Self {
             Self(.criticalStreamClosed)
         }
+
+        /// The peer terminated the inbound stream before delivering a complete request or response.
+        public static var peerTerminatedInboundStream: Self {
+            Self(.peerTerminatedInboundStream)
+        }
     }
 
     /// A location within source code.

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: .requestIncomplete,
+                    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: nil,
+                    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 {