From 588c1f12d32cc7602dda2860924494a4b54dfd7c Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Wed, 15 Jul 2026 19:23:46 +0100 Subject: [PATCH 1/9] Correctly handle inputClosed event --- Package.swift | 1 + Sources/HTTP3/HTTP3Error.swift | 6 ++ Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift | 45 ++++++++- .../NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift | 98 +++++++++++++++++++ 4 files changed, 148 insertions(+), 2 deletions(-) diff --git a/Package.swift b/Package.swift index fdac766..128d6ad 100644 --- a/Package.swift +++ b/Package.swift @@ -94,6 +94,7 @@ let package = Package( .product(name: "NIOCore", package: "swift-nio"), .product(name: "NIOConcurrencyHelpers", package: "swift-nio"), .product(name: "NIOEmbedded", package: "swift-nio"), + .product(name: "NIOQUICHelpers", package: "swift-nio-quic-helpers"), .product(name: "NIOPosix", package: "swift-nio"), .product(name: "NIOTestUtils", package: "swift-nio"), .product(name: "X509", package: "swift-certificates"), diff --git a/Sources/HTTP3/HTTP3Error.swift b/Sources/HTTP3/HTTP3Error.swift index 36546b8..15b4c90 100644 --- a/Sources/HTTP3/HTTP3Error.swift +++ b/Sources/HTTP3/HTTP3Error.swift @@ -140,6 +140,7 @@ extension HTTP3Error { case none case invalidGoawayStreamID case criticalStreamClosed + case peerTerminatedStream } public var description: String { @@ -250,6 +251,11 @@ extension HTTP3Error { public static var criticalStreamClosed: Self { Self(.criticalStreamClosed) } + + /// The peer terminated the stream before delivering a complete request or response. + public static var peerTerminatedStream: Self { + Self(.peerTerminatedStream) + } } /// A location within source code. diff --git a/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift b/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift index 84e3883..e299ae2 100644 --- a/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift +++ b/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift @@ -16,6 +16,7 @@ public import HTTP3 package import HTTPTypes public import NIOCore public import NIOHTTPTypes +internal import NIOQUICHelpers package protocol HTTPMessagePart { static func head(fields: [HTTPField]) throws(HTTP3Error) -> Self @@ -264,13 +265,16 @@ package struct HTTPMessageParsingStateMachine { package enum InputClosedAction { case returnPart(Part) + case notifyMessageIncomplete } package mutating func inputClosed() -> InputClosedAction? { switch self.state { case .awaitingHeaders: - self.state = .messageComplete - return .returnPart(.end()) + // Either no request/response head was received at all, or, in the case of clients, only interim response + // heads were received. In either case, the message is incomplete. + self.state = .failed + return .notifyMessageIncomplete case .awaitingBodyOrTrailers: self.state = .messageComplete return .returnPart(.end()) @@ -363,6 +367,18 @@ public final class HTTP3ToHTTPClientCodec: ChannelDuplexHandler { case .returnPart(let part): context.fireChannelRead(self.wrapInboundOut(part)) context.fireChannelReadComplete() + case .notifyMessageIncomplete: + // Inform the downstream application about this. + // See https://datatracker.ietf.org/doc/html/rfc9114#section-4.1.1-6. + context.fireErrorCaught( + HTTP3Error( + code: .peerTerminatedStream, + message: "Stream closed before a complete response was received", + cause: nil, + errorCode: .H3_NO_ERROR, + location: .here() + ) + ) case .none: break } @@ -384,6 +400,9 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { private var readState: HTTPMessageParsingStateMachine = .init() + /// Whether a complete response has been written. + private var completedResponse = false + public init() {} public func channelRead(context: ChannelHandlerContext, data: NIOAny) { @@ -428,6 +447,8 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { // No trailers, just close context.close(mode: .output, promise: promise) } + + self.completedResponse = true } } @@ -441,6 +462,26 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { case .returnPart(let part): context.fireChannelRead(self.wrapInboundOut(part)) context.fireChannelReadComplete() + case .notifyMessageIncomplete: + // The client terminated the stream before delivering a complete request. Per RFC 9114 § 4.1, the server + // should abort the response stream with H3_REQUEST_INCOMPLETE (unless a complete response was already + // written). + if self.completedResponse { break } + + context.triggerUserOutboundEvent( + QUICResetStreamEvent(code: QUICApplicationErrorCode(.H3_REQUEST_INCOMPLETE)), + promise: nil + ) + // Inform the downstream application about this. + context.fireErrorCaught( + HTTP3Error( + code: .peerTerminatedStream, + message: "Stream closed before a complete request was received", + cause: nil, + errorCode: .H3_REQUEST_INCOMPLETE, + location: .here() + ) + ) case .none: break } diff --git a/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift b/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift index 2dd43df..b54b033 100644 --- a/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift +++ b/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift @@ -14,10 +14,13 @@ import HTTP3 import HTTPTypes +import NIOConcurrencyHelpers import NIOCore import NIOEmbedded +import NIOExtras import NIOHTTP3 import NIOHTTPTypes +import NIOQUICHelpers import Testing struct HTTP3ToHTTPCodecTests { @@ -159,4 +162,99 @@ struct HTTP3ToHTTPCodecTests { #expect(parts[1] == .body(.init(bytes: [1, 2, 3]))) #expect(parts[2] == .body(.init(bytes: [1, 2, 3]))) } + + @Test + func serverCodecAbortsStreamWhenRequestIncompleteAndInputClosed() throws { + let codec = HTTP3ToHTTPServerCodec() + let outboundEvents = NIOLockedValueBox([]) + let outboundEventRecorder = DebugOutboundEventsHandler { event, _ in + if case .triggerUserOutboundEvent(let outboundEvent) = event { + outboundEvents.withLockedValue { $0.append(outboundEvent) } + } + } + + let eventLoop = EmbeddedEventLoop() + let errorPromise = eventLoop.makePromise(of: (any Error).self) + let errorRecorder = InboundErrorRecorder(errorPromise: errorPromise) + + let channel = EmbeddedChannel(handlers: [outboundEventRecorder, codec, errorRecorder], loop: eventLoop) + + // The client terminated the stream before even sending a request head. + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + // The server should abort its response stream with H3_REQUEST_INCOMPLETE. + let events = outboundEvents.withLockedValue { $0 } + try #require(events.count == 1) + let resetStreamEvent = try #require(events.first as? QUICResetStreamEvent) + #expect(resetStreamEvent.code == QUICApplicationErrorCode(.H3_REQUEST_INCOMPLETE)) + + // The failure should also be surfaced to the application. + let error = try errorPromise.futureResult.wait() + let h3Error = try #require(error as? HTTP3Error) + #expect(h3Error.code == .peerTerminatedStream) + #expect(h3Error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + } + + @Test + func clientCodecFiresErrorWhenResponseIncompleteAndInputClosed() throws { + let codec = HTTP3ToHTTPClientCodec() + + let eventLoop = EmbeddedEventLoop() + let errorPromise = eventLoop.makePromise(of: (any Error).self) + let errorRecorder = InboundErrorRecorder(errorPromise: errorPromise) + + let responsePartsPromise = eventLoop.makePromise(of: [HTTPResponsePart].self) + let responsePartsRecorder = InboundDataRecorder(promise: responsePartsPromise, targetCount: 2) + + let channel = EmbeddedChannel(handlers: [codec, errorRecorder, responsePartsRecorder], loop: eventLoop) + + // The client receives two interim responses, after which the server cleanly closes its send side. + try channel.writeInbound(HTTP3Frame.headers([.init(name: .status, value: "100")])) + try channel.writeInbound(HTTP3Frame.headers([.init(name: .status, value: "103")])) + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + // But the response is incomplete since the server didn't send a final response... + let parts = try responsePartsPromise.futureResult.wait() + try #require(parts.count == 2) + #expect(parts[0] == .head(.init(status: .continue))) + #expect(parts[1] == .head(.init(status: .earlyHints))) + + // so the client should surface an error to the application. + let error = try errorPromise.futureResult.wait() + let h3Error = try #require(error as? HTTP3Error) + #expect(h3Error.code == .peerTerminatedStream) + #expect(h3Error.h3ErrorCode == .H3_NO_ERROR) + } + + @Test + func serverCodecDoesNotAbortStreamWhenRequestCompleted() throws { + let codec = HTTP3ToHTTPServerCodec() + + let outboundEvents = NIOLockedValueBox([]) + let outboundEventRecorder = DebugOutboundEventsHandler { event, _ in + if case .triggerUserOutboundEvent(let outboundEvent) = event { + outboundEvents.withLockedValue { $0.append(outboundEvent) } + } + } + + let eventLoop = EmbeddedEventLoop() + let requestPartsPromise = eventLoop.makePromise(of: [HTTPRequestPart].self) + let requestPartsRecorder = InboundDataRecorder(promise: requestPartsPromise, targetCount: 2) + + let channel = EmbeddedChannel(handlers: [outboundEventRecorder, codec, requestPartsRecorder], loop: eventLoop) + print(channel.pipeline) + + // The server receives a complete request, after which the client cleanly closes its send side. + try channel.writeInbound(self.validRequestHead) + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + // Since the request was complete... + let parts = try requestPartsPromise.futureResult.wait() + try #require(parts.count == 2) + #expect(parts[0] == .head(.init(method: .get, scheme: "https", authority: "test", path: "/"))) + #expect(parts[1] == .end(nil)) + + // the stream must not be aborted. + #expect(outboundEvents.withLockedValue { $0 }.isEmpty) + } } From 91fb4bb5afb3a55040374a824d744469746a2e62 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Fri, 7 Aug 2026 15:27:36 +0100 Subject: [PATCH 2/9] Move handling of inputClosed to HTTP3StreamHandler --- Sources/HTTP3/HTTP3Error.swift | 8 +- Sources/HTTP3/HTTP3FrameValidator.swift | 154 +++++++++++++++++- Sources/HTTP3/HTTP3StreamStateMachine.swift | 33 +++- Sources/NIOHTTP3/HTTP3StreamHandler.swift | 33 +++- Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift | 45 +---- .../AsyncEndToEndTests.swift | 98 +++++++++++ .../HTTP3StreamStateMachineTests.swift | 88 +++++++++- .../HTTP3StreamHandlerTests.swift | 131 ++++++++++++++- .../NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift | 149 +++++++++-------- 9 files changed, 607 insertions(+), 132 deletions(-) diff --git a/Sources/HTTP3/HTTP3Error.swift b/Sources/HTTP3/HTTP3Error.swift index 15b4c90..53411f7 100644 --- a/Sources/HTTP3/HTTP3Error.swift +++ b/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) } } diff --git a/Sources/HTTP3/HTTP3FrameValidator.swift b/Sources/HTTP3/HTTP3FrameValidator.swift index 535fbed..d370f96 100644 --- a/Sources/HTTP3/HTTP3FrameValidator.swift +++ b/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 { diff --git a/Sources/HTTP3/HTTP3StreamStateMachine.swift b/Sources/HTTP3/HTTP3StreamStateMachine.swift index 47d7cb7..5cf2bbb 100644 --- a/Sources/HTTP3/HTTP3StreamStateMachine.swift +++ b/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) diff --git a/Sources/NIOHTTP3/HTTP3StreamHandler.swift b/Sources/NIOHTTP3/HTTP3StreamHandler.swift index 66e03f5..935b715 100644 --- a/Sources/NIOHTTP3/HTTP3StreamHandler.swift +++ b/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") diff --git a/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift b/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift index e299ae2..c2eb763 100644 --- a/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift +++ b/Sources/NIOHTTP3/HTTP3ToHTTPCodecs.swift @@ -16,7 +16,6 @@ public import HTTP3 package import HTTPTypes public import NIOCore public import NIOHTTPTypes -internal import NIOQUICHelpers package protocol HTTPMessagePart { static func head(fields: [HTTPField]) throws(HTTP3Error) -> Self @@ -265,16 +264,15 @@ package struct HTTPMessageParsingStateMachine { package enum InputClosedAction { case returnPart(Part) - case notifyMessageIncomplete } package mutating func inputClosed() -> InputClosedAction? { switch self.state { case .awaitingHeaders: - // Either no request/response head was received at all, or, in the case of clients, only interim response - // heads were received. In either case, the message is incomplete. + // The input was closed without even receiving a head part. This case is handled appropriately by + // ``HTTP3StreamHandler``, so just return `.none` here. self.state = .failed - return .notifyMessageIncomplete + return .none case .awaitingBodyOrTrailers: self.state = .messageComplete return .returnPart(.end()) @@ -367,18 +365,6 @@ public final class HTTP3ToHTTPClientCodec: ChannelDuplexHandler { case .returnPart(let part): context.fireChannelRead(self.wrapInboundOut(part)) context.fireChannelReadComplete() - case .notifyMessageIncomplete: - // Inform the downstream application about this. - // See https://datatracker.ietf.org/doc/html/rfc9114#section-4.1.1-6. - context.fireErrorCaught( - HTTP3Error( - code: .peerTerminatedStream, - message: "Stream closed before a complete response was received", - cause: nil, - errorCode: .H3_NO_ERROR, - location: .here() - ) - ) case .none: break } @@ -400,9 +386,6 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { private var readState: HTTPMessageParsingStateMachine = .init() - /// Whether a complete response has been written. - private var completedResponse = false - public init() {} public func channelRead(context: ChannelHandlerContext, data: NIOAny) { @@ -447,8 +430,6 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { // No trailers, just close context.close(mode: .output, promise: promise) } - - self.completedResponse = true } } @@ -462,26 +443,6 @@ public final class HTTP3ToHTTPServerCodec: ChannelDuplexHandler { case .returnPart(let part): context.fireChannelRead(self.wrapInboundOut(part)) context.fireChannelReadComplete() - case .notifyMessageIncomplete: - // The client terminated the stream before delivering a complete request. Per RFC 9114 § 4.1, the server - // should abort the response stream with H3_REQUEST_INCOMPLETE (unless a complete response was already - // written). - if self.completedResponse { break } - - context.triggerUserOutboundEvent( - QUICResetStreamEvent(code: QUICApplicationErrorCode(.H3_REQUEST_INCOMPLETE)), - promise: nil - ) - // Inform the downstream application about this. - context.fireErrorCaught( - HTTP3Error( - code: .peerTerminatedStream, - message: "Stream closed before a complete request was received", - cause: nil, - errorCode: .H3_REQUEST_INCOMPLETE, - location: .here() - ) - ) case .none: break } diff --git a/Tests/H3IntegrationTests/AsyncEndToEndTests.swift b/Tests/H3IntegrationTests/AsyncEndToEndTests.swift index 6e230ab..f4b7219 100644 --- a/Tests/H3IntegrationTests/AsyncEndToEndTests.swift +++ b/Tests/H3IntegrationTests/AsyncEndToEndTests.swift @@ -363,6 +363,104 @@ struct AsyncEndToEndTests { try await clientChannel.close() } + @Test(arguments: Self.standardAuthenticationConfigurations) + @available(anyAppleOS 26, *) + func streamClosesAfterInputClosedWithIncompleteRequest( + authenticationConfiguration: AuthenticationConfiguration + ) async throws { + let serverLogger = Logger(label: "Server") + let clientLogger = Logger(label: "Client") + + let credentials = try TestCertificates.makeCredentials(for: authenticationConfiguration) + let serverName: String + switch credentials { + case .rawKeys(let name, _, _, _): + serverName = name + case .certificates(let name, _, _, _): + serverName = name + } + + let (serverChannel, serverMultiplexer) = try await makeHTTP3Server( + credentials: credentials, + settings: .init(), + logger: serverLogger + ) + let (clientChannel, clientMultiplexer) = try await makeHTTP3Client( + credentials: credentials, + settings: .init(), + logger: clientLogger + ) + + try await withThrowingTaskGroup(of: Void.self) { group in + // Server + group.addTask { + var inboundConnections = serverMultiplexer.inboundConnections.makeAsyncIterator() + let inboundConnection = try #require(await inboundConnections.next()) + + var inboundStreams = inboundConnection.inboundStreams.makeAsyncIterator() + let inboundStream = try #require(await inboundStreams.next()) + + try await inboundStream.executeThenClose { inboundParts, _ in + let error = try await #require(throws: HTTP3Error.self) { + var inboundPartIterator = inboundParts.makeAsyncIterator() + // The client closes the input without sending .end(). This makes the request incomplete. + _ = try await inboundPartIterator.next() + } + + #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(error.code == .peerTerminatedInboundStream) + + // The stream should close now that we have sent a RESET_STREAM. + try await inboundStream.channel.closeFuture.get() + } + } + + // Client + group.addTask { + let clientConnection = try await clientMultiplexer.concurrencyView.createConnection( + serverName: serverName, + remoteAddress: serverChannel.localAddress!, + inboundPushStreamInitializer: { _ in + fatalError("Push streams not supported") + } + ) + + let outboundStream = try await clientConnection.concurrencyView.createRequestStream { + let streamChannel = $0.channel + return streamChannel.eventLoop.makeCompletedFuture { + try NIOAsyncChannel( + wrappingChannelSynchronously: streamChannel, + configuration: .init(isOutboundHalfClosureEnabled: true) + ) + } + } + + try await outboundStream.executeThenClose { inboundParts, outbound in + // Close the outbound side without sending a request. + outbound.finish() + + // Try to read the response. Since the server will send a RESET_STREAM for the incomplete request, we + // should see an error. + var inboundPartIterator = inboundParts.makeAsyncIterator() + let error = try await #require(throws: HTTP3Error.self) { + _ = try await inboundPartIterator.next() + } + #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(error.code == .remoteStreamError) + + // The stream should close now that we have received a RESET_STREAM. + try await outboundStream.channel.closeFuture.get() + } + } + + try await group.waitForAll() + } + + // cleanup + try await clientChannel.close() + try await serverChannel.close() + } + // MARK: - Helper Functions @available(anyAppleOS 26, *) diff --git a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift index 38fea8d..96c88cc 100644 --- a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift +++ b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift @@ -491,14 +491,92 @@ struct HTTP3StreamStateMachineTests { // MARK: Input closed @Test - func testInputClosed() { + func testInputClosedBeforeReceivingCompleteRequest() { var machine = HTTP3StreamStateMachine(streamType: .request, incoming: true, preferHuffmanEncoding: false) machine.inputClosed() let action = machine.decodeNext() - guard case .inputClosed = action else { + guard case .inputClosed(.resetStream(let error)) = action else { Issue.record("Unexpected action \(action)") return } + + expectH3ErrorEqual( + error: error, + expectedCode: .peerTerminatedInboundStream, + expectedH3ErrorCode: .H3_REQUEST_INCOMPLETE + ) + } + + @Test + func testInputClosedBeforeReceivingCompleteResponse() { + var machine = HTTP3StreamStateMachine(streamType: .request, incoming: false, preferHuffmanEncoding: false) + machine.inputClosed() + let action = machine.decodeNext() + + guard case .inputClosed(.emitErrorAndEvent(let error)) = action else { + Issue.record("Unexpected action \(action)") + return + } + + expectH3ErrorEqual( + error: error, + expectedCode: .peerTerminatedInboundStream, + expectedH3ErrorCode: .H3_NO_ERROR + ) + } + + @Test + func testInputClosedAfterReceivingCompleteRequest() { + var machine = HTTP3StreamStateMachine(streamType: .request, incoming: true, preferHuffmanEncoding: false) + + machine.buffer(.init(bytes: self.testRequestHeaderFrameBytes)) + + let action1 = machine.decodeNext() + guard case .decodeHeader(let headerToDecode) = action1 else { + Issue.record("Unexpected action \(action1)") + return + } + + machine.gotHeaderDecodeResult(self.testRequestHeaderFields, from: headerToDecode) + machine.inputClosed() + + machine.assertReturnFrame(expected: .headers(self.testRequestHeaderFields)) + let closeAction = machine.decodeNext() + + // If the input closed after receiving a complete request, the state machine should just tell us to fire the + // inputClosed event. + guard case .inputClosed(.emitEvent) = closeAction else { + Issue.record("Unexpected action \(closeAction)") + return + } + } + + @Test + func testInputClosedAfterReceivingCompleteResponse() { + var machine = HTTP3StreamStateMachine(streamType: .request, incoming: false, preferHuffmanEncoding: false) + + // Before simulating receiving a response, we must send a request. + _ = machine.writeFrame(frame: .headers(self.testRequestHeaderFields)) + machine.buffer(.init(bytes: self.testResponseHeaderFrameBytes)) + + let action1 = machine.decodeNext() + guard case .decodeHeader(let headerToDecode) = action1 else { + Issue.record("Unexpected action \(action1)") + return + } + + machine.gotHeaderDecodeResult(self.testResponseHeaderFields, from: headerToDecode) + machine.inputClosed() + + machine.assertReturnFrame(expected: .headers(self.testResponseHeaderFields)) + let closeAction = machine.decodeNext() + + // If the input closed after receiving a complete response, the state machine should just tell us to fire the + // inputClosed event. + guard case .inputClosed(.emitEvent) = closeAction else { + Issue.record("Unexpected action \(closeAction)") + return + } } @Test @@ -519,7 +597,7 @@ struct HTTP3StreamStateMachineTests { // Now the headers are returned, then the input close, then nothing else machine.assertReturnFrame(expected: .headers(self.testRequestHeaderFields)) let action2 = machine.decodeNext() - guard case .inputClosed = action2 else { + guard case .inputClosed(.emitEvent) = action2 else { Issue.record("Unexpected action \(action2)") return } @@ -544,7 +622,7 @@ struct HTTP3StreamStateMachineTests { // Now the headers are returned, then the input close, then nothing else machine.assertReturnFrame(expected: .headers(self.testRequestHeaderFields)) let action2 = machine.decodeNext() - guard case .inputClosed = action2 else { + guard case .inputClosed(.emitEvent) = action2 else { Issue.record("Unexpected action \(action2)") return } @@ -570,7 +648,7 @@ struct HTTP3StreamStateMachineTests { // Now the headers are returned, then the input close, then nothing else machine.assertReturnFrame(expected: .headers(self.testRequestHeaderFields)) let action2 = machine.decodeNext() - guard case .inputClosed = action2 else { + guard case .inputClosed(.emitEvent) = action2 else { Issue.record("Unexpected action \(action2)") return } diff --git a/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift b/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift index 880b469..16834de 100644 --- a/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift +++ b/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift @@ -495,7 +495,7 @@ struct NIOHTTP3StreamHandlerTests { } @Test - func testMoreInputAfterInputClosed() async throws { + func testMoreInputAfterInputClosed() throws { let eventLoop = EmbeddedEventLoop() let handler = HTTP3StreamHandler( @@ -523,9 +523,136 @@ struct NIOHTTP3StreamHandlerTests { // Data frame try channel.writeInbound(ByteBuffer(bytes: [0, 4, 1, 2, 3, 4])) - try await Task.sleep(for: .milliseconds(500), tolerance: .zero) // We only see the headers frame, not the data let seenFrames = recorder.getDataOnEventloop() #expect(seenFrames.count == 1) } + + @Test + func inputClosedWithIncompleteRequest() throws { + let eventLoop = EmbeddedEventLoop() + + let handler = HTTP3StreamHandler( + stateMachine: .init(streamType: .request, incoming: true, preferHuffmanEncoding: false), + streamID: 5, + streamType: .request, + qpackEncoder: self.testEncoderClosure, + qpackDecoder: { _, _ in }, + onStreamClosed: { _, _, _ in }, + onConnectionError: { Issue.record("Unexpected connection error \($0)") }, + logger: self.logger + ) + + let inboundEvents = NIOLockedValueBox<[DebugInboundEventsHandler.Event]>([]) + let inboundEventRecorder = DebugInboundEventsHandler { event, _ in + inboundEvents.withLockedValue { $0.append(event) } + } + + let outboundEvents = NIOLockedValueBox<[DebugOutboundEventsHandler.Event]>([]) + let outboundEventRecorder = DebugOutboundEventsHandler { event, _ in + outboundEvents.withLockedValue { $0.append(event) } + } + + let channel = EmbeddedChannel( + handlers: [outboundEventRecorder, handler, inboundEventRecorder], + loop: eventLoop + ) + + // Close the input before having received a complete request. + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + try handler.channelReadComplete(context: channel.pipeline.syncOperations.context(handler: handler)) + + let recordedInboundEvents = inboundEvents.withLockedValue { $0 } + let recordedOutboundEvents = outboundEvents.withLockedValue { $0 } + + try #require(recordedInboundEvents.count == 3) + #expect(recordedInboundEvents[0].isChannelRegistered) + let error = try #require(recordedInboundEvents[1].isHTTP3Error) + #expect(error.code == .peerTerminatedInboundStream) + #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(recordedInboundEvents[2].isInputClosedEvent) + + try #require(recordedOutboundEvents.count == 2) + #expect(recordedOutboundEvents[0].isChannelRegistered) + let resetStreamEvent = try #require(recordedOutboundEvents[1].isResetStreamEvent) + #expect(resetStreamEvent.code == QUICApplicationErrorCode(HTTP3ErrorCode.H3_REQUEST_INCOMPLETE)) + } + + @Test + func inputClosedWithIncompleteResponse() throws { + let eventLoop = EmbeddedEventLoop() + + let handler = HTTP3StreamHandler( + stateMachine: .init(streamType: .request, incoming: false, preferHuffmanEncoding: false), + streamID: 5, + streamType: .request, + qpackEncoder: self.testEncoderClosure, + qpackDecoder: { _, _ in }, + onStreamClosed: { _, _, _ in }, + onConnectionError: { Issue.record("Unexpected connection error \($0)") }, + logger: self.logger + ) + + let inboundEvents = NIOLockedValueBox<[DebugInboundEventsHandler.Event]>([]) + let inboundEventRecorder = DebugInboundEventsHandler { event, _ in + inboundEvents.withLockedValue { $0.append(event) } + } + let channel = EmbeddedChannel(handlers: [handler, inboundEventRecorder], loop: eventLoop) + + // Close the input before having received a complete request. + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + try handler.channelReadComplete(context: channel.pipeline.syncOperations.context(handler: handler)) + + let recordedInboundEvents = inboundEvents.withLockedValue { $0 } + + try #require(recordedInboundEvents.count == 3) + #expect(recordedInboundEvents[0].isChannelRegistered) + let error = try #require(recordedInboundEvents[1].isHTTP3Error) + #expect(error.code == .peerTerminatedInboundStream) + #expect(recordedInboundEvents[2].isInputClosedEvent) + } +} + +extension DebugInboundEventsHandler.Event { + var isInputClosedEvent: Bool { + switch self { + case .userInboundEventTriggered(let event as ChannelEvent): + return event == .inputClosed + + default: + return false + } + } + + var isHTTP3Error: HTTP3Error? { + switch self { + case .errorCaught(let error as HTTP3Error): + return error + + default: + return nil + } + } +} + +extension DebugOutboundEventsHandler.Event { + var isChannelRegistered: Bool { + switch self { + case .register: + return true + + default: + return false + } + } + + var isResetStreamEvent: NIOQUICHelpers.QUICResetStreamEvent? { + switch self { + case .triggerUserOutboundEvent(let event as NIOQUICHelpers.QUICResetStreamEvent): + return event + + default: + return nil + } + } } diff --git a/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift b/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift index b54b033..8d2a292 100644 --- a/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift +++ b/Tests/NIOHTTP3Tests/HTTP3ToHTTPCodecTests.swift @@ -20,7 +20,6 @@ import NIOEmbedded import NIOExtras import NIOHTTP3 import NIOHTTPTypes -import NIOQUICHelpers import Testing struct HTTP3ToHTTPCodecTests { @@ -31,10 +30,14 @@ struct HTTP3ToHTTPCodecTests { .init(name: .path, value: "/"), ]) - private let validResponseHead: HTTP3Frame = .headers([ + private let validFinalResponseHead: HTTP3Frame = .headers([ .init(name: .status, value: "200") ]) + private let validInterimResponseHead: HTTP3Frame = .headers([ + .init(name: .status, value: "103") + ]) + @Test func testClientCodecWrite() throws { let handler = HTTP3ToHTTPClientCodec() @@ -89,7 +92,7 @@ struct HTTP3ToHTTPCodecTests { let recorder = InboundDataRecorder(promise: partsPromise, targetCount: 3) let channel = EmbeddedChannel(handlers: [handler, recorder], loop: eventLoop) - try channel.writeInbound(self.validResponseHead) + try channel.writeInbound(self.validFinalResponseHead) try channel.writeInbound(HTTP3Frame.data(.init(bytes: [1, 2, 3]))) try channel.writeInbound(HTTP3Frame.data(.init(bytes: [1, 2, 3]))) @@ -164,97 +167,105 @@ struct HTTP3ToHTTPCodecTests { } @Test - func serverCodecAbortsStreamWhenRequestIncompleteAndInputClosed() throws { - let codec = HTTP3ToHTTPServerCodec() - let outboundEvents = NIOLockedValueBox([]) - let outboundEventRecorder = DebugOutboundEventsHandler { event, _ in - if case .triggerUserOutboundEvent(let outboundEvent) = event { - outboundEvents.withLockedValue { $0.append(outboundEvent) } - } - } - + func serverCodecGeneratesEndPartForCompleteRequestUponInputClosed() throws { let eventLoop = EmbeddedEventLoop() - let errorPromise = eventLoop.makePromise(of: (any Error).self) - let errorRecorder = InboundErrorRecorder(errorPromise: errorPromise) + let partsPromise = eventLoop.makePromise(of: [HTTPRequestPart].self) + let dataRecorder = InboundDataRecorder(promise: partsPromise, targetCount: 2) - let channel = EmbeddedChannel(handlers: [outboundEventRecorder, codec, errorRecorder], loop: eventLoop) + let codec = HTTP3ToHTTPServerCodec() - // The client terminated the stream before even sending a request head. + let channel = EmbeddedChannel(handlers: [codec, dataRecorder], loop: eventLoop) + + // Write a request head and then close the input side. + try channel.writeInbound(self.validRequestHead) channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) - // The server should abort its response stream with H3_REQUEST_INCOMPLETE. - let events = outboundEvents.withLockedValue { $0 } - try #require(events.count == 1) - let resetStreamEvent = try #require(events.first as? QUICResetStreamEvent) - #expect(resetStreamEvent.code == QUICApplicationErrorCode(.H3_REQUEST_INCOMPLETE)) - - // The failure should also be surfaced to the application. - let error = try errorPromise.futureResult.wait() - let h3Error = try #require(error as? HTTP3Error) - #expect(h3Error.code == .peerTerminatedStream) - #expect(h3Error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + let parts = try partsPromise.futureResult.wait() + try #require(parts.count == 2) + #expect(parts[0] == .head(.init(method: .get, scheme: "https", authority: "test", path: "/"))) + #expect(parts[1] == .end(nil)) } @Test - func clientCodecFiresErrorWhenResponseIncompleteAndInputClosed() throws { - let codec = HTTP3ToHTTPClientCodec() - + func clientCodecGeneratesEndPartForCompleteResponseUponInputClosed() throws { let eventLoop = EmbeddedEventLoop() - let errorPromise = eventLoop.makePromise(of: (any Error).self) - let errorRecorder = InboundErrorRecorder(errorPromise: errorPromise) + let partsPromise = eventLoop.makePromise(of: [HTTPResponsePart].self) + let dataRecorder = InboundDataRecorder(promise: partsPromise, targetCount: 2) - let responsePartsPromise = eventLoop.makePromise(of: [HTTPResponsePart].self) - let responsePartsRecorder = InboundDataRecorder(promise: responsePartsPromise, targetCount: 2) + let codec = HTTP3ToHTTPClientCodec() - let channel = EmbeddedChannel(handlers: [codec, errorRecorder, responsePartsRecorder], loop: eventLoop) + let channel = EmbeddedChannel(handlers: [codec, dataRecorder], loop: eventLoop) - // The client receives two interim responses, after which the server cleanly closes its send side. - try channel.writeInbound(HTTP3Frame.headers([.init(name: .status, value: "100")])) - try channel.writeInbound(HTTP3Frame.headers([.init(name: .status, value: "103")])) + // Write a response head and then close the input side. + try channel.writeInbound(self.validFinalResponseHead) channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) - // But the response is incomplete since the server didn't send a final response... - let parts = try responsePartsPromise.futureResult.wait() + let parts = try partsPromise.futureResult.wait() try #require(parts.count == 2) - #expect(parts[0] == .head(.init(status: .continue))) - #expect(parts[1] == .head(.init(status: .earlyHints))) - - // so the client should surface an error to the application. - let error = try errorPromise.futureResult.wait() - let h3Error = try #require(error as? HTTP3Error) - #expect(h3Error.code == .peerTerminatedStream) - #expect(h3Error.h3ErrorCode == .H3_NO_ERROR) + #expect(parts[0] == .head(.init(status: .ok))) + #expect(parts[1] == .end(nil)) } @Test - func serverCodecDoesNotAbortStreamWhenRequestCompleted() throws { + func serverCodecDoesNotFireChannelReadWhenNoRequest() throws { + let eventLoop = EmbeddedEventLoop() + let promise = eventLoop.makePromise(of: [Void].self) + let inboundEvents = NIOLockedValueBox<[DebugInboundEventsHandler.Event]>([]) + let eventRecorder = DebugInboundEventsHandler { event, context in + inboundEvents.withLockedValue { $0.append(event) } + } + let codec = HTTP3ToHTTPServerCodec() - let outboundEvents = NIOLockedValueBox([]) - let outboundEventRecorder = DebugOutboundEventsHandler { event, _ in - if case .triggerUserOutboundEvent(let outboundEvent) = event { - outboundEvents.withLockedValue { $0.append(outboundEvent) } - } - } + let channel = EmbeddedChannel(handlers: [codec, eventRecorder], loop: eventLoop) + + // The client terminated the stream before even sending a request head. + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + // We expect the server codec to not fire any channelReads if no request parts were sent before the input closed; + // `HTTP3StreamHandler` handles this case by resetting the stream and firing an error down the pipeline. + let events = inboundEvents.withLockedValue { $0 } + #expect(events.count == 2) + #expect(events[0].isChannelRegistered) + #expect(events[1].isInputClosedEvent) + + // Clean up the promise as it never gets fulfilled. + promise.succeed([Void]()) + } + @Test(arguments: [true, false]) + func clientCodecDoesNotEmitEndWhenResponseIncomplete(includeInterimResponses: Bool) throws { let eventLoop = EmbeddedEventLoop() - let requestPartsPromise = eventLoop.makePromise(of: [HTTPRequestPart].self) - let requestPartsRecorder = InboundDataRecorder(promise: requestPartsPromise, targetCount: 2) + let promise = eventLoop.makePromise(of: [HTTPResponsePart].self) + let dataRecorder = InboundDataRecorder(promise: promise, targetCount: includeInterimResponses ? 2 : 0) - let channel = EmbeddedChannel(handlers: [outboundEventRecorder, codec, requestPartsRecorder], loop: eventLoop) - print(channel.pipeline) + let codec = HTTP3ToHTTPClientCodec() - // The server receives a complete request, after which the client cleanly closes its send side. - try channel.writeInbound(self.validRequestHead) - channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + let channel = EmbeddedChannel(handlers: [codec, dataRecorder], loop: eventLoop) - // Since the request was complete... - let parts = try requestPartsPromise.futureResult.wait() - try #require(parts.count == 2) - #expect(parts[0] == .head(.init(method: .get, scheme: "https", authority: "test", path: "/"))) - #expect(parts[1] == .end(nil)) + if includeInterimResponses { + // Receiving interim response head(s) does not mean the response is complete; the response is only + // considered complete once a *final* response head is received. + try channel.writeInbound(self.validInterimResponseHead) + try channel.writeInbound(self.validInterimResponseHead) + } - // the stream must not be aborted. - #expect(outboundEvents.withLockedValue { $0 }.isEmpty) + // The server terminated the stream before sending a final response head. + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + // We expect the client codec to not generate an end part if the response is incomplete when the input closed; + // `HTTP3StreamHandler` handles this case by firing an error down the pipeline. + if includeInterimResponses { + let parts = try promise.futureResult.wait() + try #require(parts.count == 2) + #expect(parts[0] == .head(.init(status: .earlyHints))) + #expect(parts[1] == .head(.init(status: .earlyHints))) + } else { + // We expect the client codec to not deliver anything when nothing was sent before the input closed. + #expect(dataRecorder.getDataOnEventloop().count == 0) + + // Clean up the promise as it never gets fulfilled when `includeInterimResponse` == `false`. + promise.succeed([HTTPResponsePart]()) + } } } From 334d01aff96b7b1730aa691b5224aee17f1eaebe Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Fri, 7 Aug 2026 15:29:55 +0100 Subject: [PATCH 3/9] Adopt new error code API --- Sources/HTTP3/HTTP3FrameValidator.swift | 4 ++-- Sources/NIOHTTP3/HTTP3StreamHandler.swift | 2 +- Tests/H3IntegrationTests/AsyncEndToEndTests.swift | 4 ++-- Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift | 8 ++------ Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift | 4 ++-- 5 files changed, 9 insertions(+), 13 deletions(-) diff --git a/Sources/HTTP3/HTTP3FrameValidator.swift b/Sources/HTTP3/HTTP3FrameValidator.swift index b8f56ff..9228906 100644 --- a/Sources/HTTP3/HTTP3FrameValidator.swift +++ b/Sources/HTTP3/HTTP3FrameValidator.swift @@ -422,7 +422,7 @@ package enum HTTP3FrameValidator: ~Copyable { code: .peerTerminatedInboundStream, message: "Inbound request stream closed before a complete request was received", cause: nil, - errorCode: .H3_REQUEST_INCOMPLETE, + errorCode: .requestIncomplete, location: .here() ) ) @@ -445,7 +445,7 @@ package enum HTTP3FrameValidator: ~Copyable { code: .peerTerminatedInboundStream, message: "Inbound response stream closed before a complete response was received", cause: nil, - errorCode: .H3_NO_ERROR, + errorCode: .noError, location: .here() ) ) diff --git a/Sources/NIOHTTP3/HTTP3StreamHandler.swift b/Sources/NIOHTTP3/HTTP3StreamHandler.swift index 9e0c6e1..b5ee7c7 100644 --- a/Sources/NIOHTTP3/HTTP3StreamHandler.swift +++ b/Sources/NIOHTTP3/HTTP3StreamHandler.swift @@ -297,7 +297,7 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler { case .resetStream(let error): context.triggerUserOutboundEvent( - QUICResetStreamEvent(code: QUICApplicationErrorCode(error.h3ErrorCode ?? .H3_NO_ERROR)), + QUICResetStreamEvent(code: QUICApplicationErrorCode(error.h3ErrorCode ?? .noError)), promise: nil ) context.fireErrorCaught(error) diff --git a/Tests/H3IntegrationTests/AsyncEndToEndTests.swift b/Tests/H3IntegrationTests/AsyncEndToEndTests.swift index 269dd9c..7992b50 100644 --- a/Tests/H3IntegrationTests/AsyncEndToEndTests.swift +++ b/Tests/H3IntegrationTests/AsyncEndToEndTests.swift @@ -407,7 +407,7 @@ struct AsyncEndToEndTests { _ = try await inboundPartIterator.next() } - #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(error.h3ErrorCode == .requestIncomplete) #expect(error.code == .peerTerminatedInboundStream) // The stream should close now that we have sent a RESET_STREAM. @@ -445,7 +445,7 @@ struct AsyncEndToEndTests { let error = try await #require(throws: HTTP3Error.self) { _ = try await inboundPartIterator.next() } - #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(error.h3ErrorCode == .requestIncomplete) #expect(error.code == .remoteStreamError) // The stream should close now that we have received a RESET_STREAM. diff --git a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift index 303ecbd..3055fe4 100644 --- a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift +++ b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift @@ -503,7 +503,7 @@ struct HTTP3StreamStateMachineTests { expectH3ErrorEqual( error: error, expectedCode: .peerTerminatedInboundStream, - expectedH3ErrorCode: .H3_REQUEST_INCOMPLETE + expectedH3ErrorCode: .requestIncomplete ) } @@ -518,11 +518,7 @@ struct HTTP3StreamStateMachineTests { return } - expectH3ErrorEqual( - error: error, - expectedCode: .peerTerminatedInboundStream, - expectedH3ErrorCode: .H3_NO_ERROR - ) + expectH3ErrorEqual(error: error, expectedCode: .peerTerminatedInboundStream, expectedH3ErrorCode: .noError) } @Test diff --git a/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift b/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift index 6fd4048..7e229ef 100644 --- a/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift +++ b/Tests/NIOHTTP3Tests/HTTP3StreamHandlerTests.swift @@ -569,13 +569,13 @@ struct NIOHTTP3StreamHandlerTests { #expect(recordedInboundEvents[0].isChannelRegistered) let error = try #require(recordedInboundEvents[1].isHTTP3Error) #expect(error.code == .peerTerminatedInboundStream) - #expect(error.h3ErrorCode == .H3_REQUEST_INCOMPLETE) + #expect(error.h3ErrorCode == .requestIncomplete) #expect(recordedInboundEvents[2].isInputClosedEvent) try #require(recordedOutboundEvents.count == 2) #expect(recordedOutboundEvents[0].isChannelRegistered) let resetStreamEvent = try #require(recordedOutboundEvents[1].isResetStreamEvent) - #expect(resetStreamEvent.code == QUICApplicationErrorCode(HTTP3ErrorCode.H3_REQUEST_INCOMPLETE)) + #expect(resetStreamEvent.code == QUICApplicationErrorCode(HTTP3ErrorCode.requestIncomplete)) } @Test From 12f344f1cd1b1fe64e9a111242369d11ce4e87d5 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 13:19:22 +0100 Subject: [PATCH 4/9] Replace noError with nil --- Sources/HTTP3/HTTP3FrameValidator.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Sources/HTTP3/HTTP3FrameValidator.swift b/Sources/HTTP3/HTTP3FrameValidator.swift index 9228906..ae05f0f 100644 --- a/Sources/HTTP3/HTTP3FrameValidator.swift +++ b/Sources/HTTP3/HTTP3FrameValidator.swift @@ -445,7 +445,7 @@ package enum HTTP3FrameValidator: ~Copyable { code: .peerTerminatedInboundStream, message: "Inbound response stream closed before a complete response was received", cause: nil, - errorCode: .noError, + errorCode: nil, location: .here() ) ) From 5dbe4baba007bdc28990fcba443c9e20902840f8 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 13:37:41 +0100 Subject: [PATCH 5/9] Handle inputClosed differently on channelReadComplete and channelInactive --- Sources/NIOHTTP3/HTTP3StreamHandler.swift | 55 ++++++++++++----------- 1 file changed, 30 insertions(+), 25 deletions(-) diff --git a/Sources/NIOHTTP3/HTTP3StreamHandler.swift b/Sources/NIOHTTP3/HTTP3StreamHandler.swift index b5ee7c7..f7858a0 100644 --- a/Sources/NIOHTTP3/HTTP3StreamHandler.swift +++ b/Sources/NIOHTTP3/HTTP3StreamHandler.swift @@ -117,7 +117,20 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler { loop: for action in actionBuffer { switch action { case .inputClosed(let inputClosedAction): - self.handleInputClosed(action: inputClosedAction, context: context) + switch inputClosedAction { + case .emitEvent: + context.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + case .emitErrorAndEvent(let error): + context.fireErrorCaught(error) + context.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + case .resetStream(let error): + // The channel is already inactive, so we cannot send a RESET_STREAM. Just emit the error and + // event downstream. + context.fireErrorCaught(error) + context.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + } case .returnFrame(let frame): context.fireChannelRead(wrapInboundOut(frame)) didFireChannelRead = true @@ -168,7 +181,22 @@ package final class HTTP3StreamHandler: ChannelDuplexHandler { let action = self.stateMachine.decodeNext() switch action { case .inputClosed(let inputClosedAction): - self.handleInputClosed(action: inputClosedAction, context: context) + switch inputClosedAction { + 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 ?? .noError)), + promise: nil + ) + context.fireErrorCaught(error) + context.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + } case .needMoreBytes, .alreadyClosed, .previousError: break decodeLoop case .callAgain: @@ -282,29 +310,6 @@ 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 ?? .noError)), - promise: nil - ) - context.fireErrorCaught(error) - context.fireUserInboundEventTriggered(ChannelEvent.inputClosed) - } - } - package func errorCaught(context: ChannelHandlerContext, error: any Error) { switch error { case let error as QUICStreamResetError: From 153804e12f85957271885c133a77a0840eac8b74 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 14:37:42 +0100 Subject: [PATCH 6/9] Keep a reference to readState in the previousError case --- Sources/HTTP3/HTTP3StreamStateMachine.swift | 57 +++++++++++++++------ 1 file changed, 40 insertions(+), 17 deletions(-) diff --git a/Sources/HTTP3/HTTP3StreamStateMachine.swift b/Sources/HTTP3/HTTP3StreamStateMachine.swift index 8e0b0a8..a12305d 100644 --- a/Sources/HTTP3/HTTP3StreamStateMachine.swift +++ b/Sources/HTTP3/HTTP3StreamStateMachine.swift @@ -435,15 +435,25 @@ package struct HTTP3StreamStateMachine: ~Copyable { case idle(Idle) /// We previously hit an error, and now can't do anything. - case previousError(HTTP3Error) + case previousError(PreviousErrorState) /// The stream is closed. case finished + struct Idle: ~Copyable { var validator: HTTP3FrameValidator var readState: ReadState var writeState: WriteState } + + /// The state contained in ``State/previousError(_:)``. + struct PreviousErrorState: ~Copyable { + /// The error that was reached. This is reported back on any subsequent operation. + var error: HTTP3Error + + /// The read state when the error occurred. + var readState: ReadState + } } private let state: State @@ -500,10 +510,10 @@ package struct HTTP3StreamStateMachine: ~Copyable { return .encodeHeaders(fields) } case .emitStreamError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .wouldBeStreamError(error) case .emitConnectionError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .wouldBeConnectionError(error) case .previousError: self = .init(state: .idle(idleState)) @@ -546,8 +556,9 @@ package struct HTTP3StreamStateMachine: ~Copyable { // But if we do, just drop it, nobody is waiting for it now. self = .init(state: .finished) return .alreadyClosed - case .previousError(let error): - self = .init(state: .previousError(error)) + case .previousError(let errorState): + let error = errorState.error + self = .init(state: .previousError(errorState)) return .previousError(error) } } @@ -616,10 +627,10 @@ package struct HTTP3StreamStateMachine: ~Copyable { self = .init(state: .idle(idleState)) return .returnFrame(validatedFrame) case .emitStreamError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitStreamError(error) case .emitConnectionError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitConnectionError(error) case .previousError: self = .init(state: .idle(idleState)) @@ -629,7 +640,7 @@ package struct HTTP3StreamStateMachine: ~Copyable { let validationResult = idleState.validator.processInboundUnknownFrame() switch validationResult { case .emitConnectionError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitConnectionError(error) case .dropFrame: self = .init(state: .idle(idleState)) @@ -641,10 +652,10 @@ package struct HTTP3StreamStateMachine: ~Copyable { return .previousError } case .emitConnectionError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitConnectionError(error) case .emitStreamError(let error): - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitStreamError(error) case .decodeHeader(let partialHeader): self = .init(state: .idle(idleState)) @@ -667,7 +678,7 @@ package struct HTTP3StreamStateMachine: ~Copyable { return .inputClosed(.emitErrorAndEvent(error)) case .resetStream(let error): - self = .init(state: .idle(idleState)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .inputClosed(.resetStream(error)) } } @@ -750,9 +761,9 @@ package struct HTTP3StreamStateMachine: ~Copyable { ) } switch consume self.state { - case .idle: + case .idle(let idleState): let error = remoteStreamError(errorCode: errorCodeValue, location: .here()) - self = .init(state: .previousError(error)) + self = .init(state: .previousError(.init(error: error, readState: idleState.readState))) return .emitStreamError(error) case .previousError(let previousError): // ignore the new error because we already are in an error state @@ -777,13 +788,25 @@ package struct HTTP3StreamStateMachine: ~Copyable { case .idle(let idle): let finishState = idle.readState.closed() self = .init(state: .finished) + switch finishState { - case .sawEOF: return .streamClosed(seenEOF: true) - case .noEOF: return .streamClosed(seenEOF: false) + case .sawEOF: + return .streamClosed(seenEOF: true) + + case .noEOF: + return .streamClosed(seenEOF: false) } - case .previousError: + case .previousError(let errorState): + let finishState = errorState.readState.closed() self = .init(state: .finished) - return .streamClosed(seenEOF: false) + + switch finishState { + case .sawEOF: + return .streamClosed(seenEOF: true) + + case .noEOF: + return .streamClosed(seenEOF: false) + } case .finished: fatalError("Finished called twice") } From 9c9d11e74ff1949995620c7cf89be3ca03aa5b78 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 14:39:27 +0100 Subject: [PATCH 7/9] Fix state machine transition --- Sources/HTTP3/HTTP3StreamStateMachine.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Sources/HTTP3/HTTP3StreamStateMachine.swift b/Sources/HTTP3/HTTP3StreamStateMachine.swift index a12305d..a145fd2 100644 --- a/Sources/HTTP3/HTTP3StreamStateMachine.swift +++ b/Sources/HTTP3/HTTP3StreamStateMachine.swift @@ -187,7 +187,7 @@ package struct HTTP3StreamStateMachine: ~Copyable { self = .init(state: .idle(.init(decoder: bufferState.decoder, seenEOF: bufferState.seenEOF))) return .returnFrame(bufferState.frame) case .headerDecodeError(let error): - self = .init(state: .inputClosed) + self = .init(state: .headerDecodeError(error)) return .emitStreamError(error.error) case .inputClosed: self = .init(state: .inputClosed) From dc8d8ee00653ae8cd71d2b22ce5afe3d748e7888 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 14:40:17 +0100 Subject: [PATCH 8/9] Fix failing test --- Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift index 3055fe4..0a2eb5e 100644 --- a/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift +++ b/Tests/HTTP3Tests/HTTP3StreamStateMachineTests.swift @@ -518,7 +518,7 @@ struct HTTP3StreamStateMachineTests { return } - expectH3ErrorEqual(error: error, expectedCode: .peerTerminatedInboundStream, expectedH3ErrorCode: .noError) + expectH3ErrorEqual(error: error, expectedCode: .peerTerminatedInboundStream, expectedH3ErrorCode: nil) } @Test From a42abce432a42afe7e153106f38709655a1efa96 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 10 Aug 2026 14:48:14 +0100 Subject: [PATCH 9/9] Remove redundant import --- Package.swift | 1 - 1 file changed, 1 deletion(-) diff --git a/Package.swift b/Package.swift index 3aa79e8..e328b30 100644 --- a/Package.swift +++ b/Package.swift @@ -94,7 +94,6 @@ let package = Package( .product(name: "NIOCore", package: "swift-nio"), .product(name: "NIOConcurrencyHelpers", package: "swift-nio"), .product(name: "NIOEmbedded", package: "swift-nio"), - .product(name: "NIOQUICHelpers", package: "swift-nio-quic-helpers"), .product(name: "NIOPosix", package: "swift-nio"), .product(name: "NIOTestUtils", package: "swift-nio"), .product(name: "X509", package: "swift-certificates"),