From 14becc70f3b2b703f8c700f983bf2045fe0ed7a4 Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Fri, 17 Jul 2026 12:41:01 +0100 Subject: [PATCH] Add reset stream API --- .../NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift | 1 + .../NIOHTTPServer/NIOHTTPServer+HTTP3.swift | 6 +- .../NIOHTTPServer+SecureUpgrade.swift | 38 +++++-- .../NIOHTTPServer+StreamReset.swift | 100 ++++++++++++++++++ Sources/NIOHTTPServer/NIOHTTPServer.swift | 7 +- .../NIOHTTPServerResponseSender.swift | 61 ++++++++++- .../NIOHTTPServerResponseSenderTests.swift | 12 ++- .../NIOHTTPServerWriterTests.swift | 18 +++- .../NegotiatedClientConnection.swift | 14 +++ 9 files changed, 241 insertions(+), 16 deletions(-) create mode 100644 Sources/NIOHTTPServer/NIOHTTPServer+StreamReset.swift diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 57a98b2..ce722ac 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -215,6 +215,7 @@ extension NIOHTTPServer { request: httpRequest, iterator: iterator, outbound: outbound, + streamReset: .unavailable, handler: handler, context: context ) diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift index b531fc8..b9eec03 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift @@ -96,7 +96,11 @@ extension NIOHTTPServer { await withDiscardingTaskGroup { streamGroup in for await streamChannel in connection.inboundStreams { streamGroup.addTask { - await self.handleStreamChannel(channel: streamChannel, handler: handler, context: context) + await self.handleStreamChannel( + channel: streamChannel, + handler: handler, + context: context + ) } } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index 5a5bbc4..974f7b7 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -185,7 +185,11 @@ extension NIOHTTPServer { do { for try await streamChannel in multiplexer.inbound { streamGroup.addTask { - await self.handleStreamChannel(channel: streamChannel, handler: handler, context: context) + await self.handleStreamChannel( + channel: streamChannel, + handler: handler, + context: context + ) } } } catch { @@ -371,6 +375,9 @@ extension NIOHTTPServer { } /// Handles a stream channel, which carries exactly one request per stream. + /// + /// Used only for HTTP/2 and HTTP/3, which have per-request streams; HTTP/1.1 is served by + /// ``handleHTTP1RequestLoop(inbound:outbound:handler:context:)``. func handleStreamChannel( channel: NIOAsyncChannel, handler: Handler, @@ -390,19 +397,38 @@ extension NIOHTTPServer { return } + let streamReset: NIOHTTPServer.StreamReset + switch context.httpVersion { + case .http2: + streamReset = .http2(.init(channel: channel.channel)) + + #if HTTP3 + case .http3: + streamReset = .http3(.init(channel: channel.channel)) + #endif // HTTP3 + + case .http1_1, .plaintextHTTP1_1: + preconditionFailure("handleStreamChannel only serves HTTP/2 and HTTP/3 streams") + } + _ = try await self.invokeHandler( request: httpRequest, iterator: iterator, outbound: outbound, + streamReset: streamReset, handler: handler, context: context ) - // TODO: handle other state scenarios. - // For example, if we didn't finish reading but we wrote back a response, we - // should send a RST_STREAM with NO_ERROR set. If we finished reading but we - // didn't write back a response, then RST_STREAM is also likely appropriate but - // unclear about the error. + // TODO: When the handler concludes the response without consuming the full request body, the request + // half of the stream is left open. Ideally we would send RST_STREAM(NO_ERROR) to tell the client to + // stop sending the request body — but only when the client has *not* already closed its half (i.e. we + // have not observed END_STREAM on the inbound side). `finishedReading` cannot distinguish these cases: + // it is `false` both when the client still has body to send *and* when the client already ended the + // stream but the handler simply never read it (e.g. a bodyless GET answered without reading). Sending + // RST_STREAM in the latter case would reset an already-closed stream. Doing this correctly requires + // threading the observed inbound END_STREAM state out of the reader / `nextRequestHead`; deferred to a + // follow-up. // Finish the outbound and wait on the close future to make sure all pending // writes are actually written. diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+StreamReset.swift b/Sources/NIOHTTPServer/NIOHTTPServer+StreamReset.swift new file mode 100644 index 0000000..6adeab6 --- /dev/null +++ b/Sources/NIOHTTPServer/NIOHTTPServer+StreamReset.swift @@ -0,0 +1,100 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift HTTP Server open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import NIOCore +public import NIOHTTP2 +import NIOHTTPTypes +import NIOHTTPTypesHTTP2 + +#if HTTP3 +import NIOQUICHelpers +#endif // HTTP3 + +@available(anyAppleOS 26.0, *) +extension NIOHTTPServer { + /// The protocol-specific surface for resetting the stream carrying a request. + /// + /// Stream resets only exists over HTTP/2 and HTTP/3. The only abrupt tear-down mechanism available + /// for HTTP/1.1 is closing the connection. + @nonexhaustive + public enum StreamReset: ~Copyable { + /// The protocol has no per-stream coded reset (for example HTTP/1.1). + /// + /// There is nothing to reset with a code here. Returning from the handler without concluding the response + /// aborts the exchange and the connection is closed. + case unavailable + + /// The connection is HTTP/2; ``HTTP2StreamReset`` sends a `RST_STREAM`. + case http2(HTTP2StreamReset) + + #if HTTP3 + /// The connection is HTTP/3; ``HTTP3StreamReset`` sends a QUIC `RESET_STREAM`. + case http3(HTTP3StreamReset) + #endif // HTTP3 + } + + /// Resets an HTTP/2 stream by sending a `RST_STREAM` frame with a chosen error code. + public struct HTTP2StreamReset: ~Copyable { + private let channel: any Channel + + init(channel: any Channel) { + self.channel = channel + } + + /// Sends a `RST_STREAM` frame for this stream with the provided error code. + /// + /// - Parameter code: The `RST_STREAM` error code to send. + public consuming func reset(code: HTTP2ErrorCode) { + // The `HTTP2FramePayloadToHTTPServerCodec` on the stream channel translates this event into an `RST_STREAM` + // frame. + self.channel.triggerUserOutboundEvent( + NIOHTTP2FramePayloadToHTTPEvent.reset(code: code), + promise: nil + ) + } + } + + #if HTTP3 + /// Resets an HTTP/3 stream by sending a QUIC `RESET_STREAM` frame with a chosen error code. + public struct HTTP3StreamReset: ~Copyable { + private let channel: any Channel + + init(channel: any Channel) { + self.channel = channel + } + + /// Sends a QUIC `RESET_STREAM` frame for this stream with the provided error code. + /// + /// - Parameter code: The QUIC application error code to send. It must be a valid application error code that is + /// less than 2^62 (the maximum QUIC variable-length integer value). If the code is out of range, the stream + /// is not reset. + public consuming func reset(code: UInt64) { + guard let resetCode = QUICApplicationErrorCode(code) else { return } + + self.channel.triggerUserOutboundEvent(QUICResetStreamEvent(code: resetCode), promise: nil) + } + } + #endif // HTTP3 +} + +@available(*, unavailable) +extension NIOHTTPServer.StreamReset: Sendable {} + +@available(*, unavailable) +extension NIOHTTPServer.HTTP2StreamReset: Sendable {} + +#if HTTP3 +@available(*, unavailable) +extension NIOHTTPServer.HTTP3StreamReset: Sendable {} +#endif // HTTP3 diff --git a/Sources/NIOHTTPServer/NIOHTTPServer.swift b/Sources/NIOHTTPServer/NIOHTTPServer.swift index 37c0bf2..9f6ca72 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer.swift @@ -332,6 +332,7 @@ public struct NIOHTTPServer: HTTPServer { request: HTTPRequest, iterator: consuming sending NIOAsyncChannelInboundStream.AsyncIterator, outbound: NIOAsyncChannelOutboundWriter, + streamReset: consuming sending NIOHTTPServer.StreamReset, handler: Handler, context: ConnectionContext ) async throws -> NIOAsyncChannelInboundStream.AsyncIterator? @@ -350,7 +351,11 @@ public struct NIOHTTPServer: HTTPServer { reader: Reader( readerState: readerState ), - responseSender: ResponseSender(writer: outbound, writerState: writerState) + responseSender: ResponseSender( + writer: outbound, + writerState: writerState, + streamReset: streamReset + ) ) } catch { logger.error("Error thrown while handling request: \(error)") diff --git a/Sources/NIOHTTPServer/NIOHTTPServerResponseSender.swift b/Sources/NIOHTTPServer/NIOHTTPServerResponseSender.swift index 2e7d898..09805bd 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServerResponseSender.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServerResponseSender.swift @@ -21,6 +21,7 @@ extension NIOHTTPServer { public struct ResponseSender: HTTPResponseSender, ~Copyable { let writer: NIOAsyncChannelOutboundWriter let writerState: WriterState + let streamReset: NIOHTTPServer.StreamReset public mutating func sendInformational(_ response: HTTPResponse) async throws { precondition(response.status.kind == .informational) @@ -30,7 +31,23 @@ extension NIOHTTPServer { public consuming func send(_ response: HTTPResponse) async throws -> Writer { precondition(response.status.kind != .informational) try await self.writer.write(.head(response)) - return Writer(writer: self.writer, writerState: self.writerState) + return Writer( + writer: self.writer, + writerState: self.writerState, + streamReset: self.streamReset + ) + } + + /// Abandons the response and resets the stream carrying this request. + /// + /// Call this instead of ``send(_:)`` when the request should be aborted before any response head is sent. This + /// consumes the sender, so no response can be sent afterwards. + /// + /// - Parameter body: A closure that is provided a ``NIOHTTPServer/StreamReset`` instance from which the request + /// stream can be reset with a transport-specific error code. + public consuming func reset(_ body: (consuming NIOHTTPServer.StreamReset) throws -> Void) throws { + self.writerState.markReset(self.streamReset) + return try body(self.streamReset) } } } @@ -43,6 +60,30 @@ extension NIOHTTPServer.ResponseSender { } let wrapped: Mutex = .init(.init()) + + /// Records that the handler chose to reset the stream instead of + /// concluding the response normally. + /// + /// On HTTP/2 (`RST_STREAM`) and HTTP/3 (`RESET_STREAM`) the coded reset is + /// itself a clean conclusion of the exchange, so the response is marked + /// as concluded to avoid an erroneous "did not conclude the response" + /// teardown. When no coded reset is available (HTTP/1.1) there is nothing + /// to send: leaving the response unconcluded is deliberate, so the + /// connection is torn down. + func markReset(_ streamReset: borrowing NIOHTTPServer.StreamReset) { + switch streamReset { + case .unavailable: + () + + case .http2: + self.wrapped.withLock { $0.finishedWriting = true } + + #if HTTP3 + case .http3: + self.wrapped.withLock { $0.finishedWriting = true } + #endif // HTTP3 + } + } } public struct Writer: CallerAsyncWriter, ~Copyable { @@ -57,6 +98,8 @@ extension NIOHTTPServer.ResponseSender { let writerState: WriterState + let streamReset: NIOHTTPServer.StreamReset + public mutating func write( buffer: inout some RangeReplaceableContainer & ~Copyable ) async throws(WriteFailure) { @@ -73,7 +116,7 @@ extension NIOHTTPServer.ResponseSender { if span.isEmpty { done = true } else { - byteBuffer.writeBytes(span.span.bytes) + unsafe byteBuffer.writeBytes(span.span.bytes) } } @@ -98,7 +141,7 @@ extension NIOHTTPServer.ResponseSender { if span.isEmpty { done = true } else { - byteBuffer.writeBytes(span.span.bytes) + unsafe byteBuffer.writeBytes(span.span.bytes) } } @@ -107,6 +150,18 @@ extension NIOHTTPServer.ResponseSender { try await self.writer.write(.end(finalElement)) self.writerState.wrapped.withLock { $0.finishedWriting = true } } + + /// Abandons the in-flight response and resets the stream carrying this request. + /// + /// Call this instead of ``finish(buffer:finalElement:)`` when a response that has already started must be + /// aborted. This consumes the writer, so no further body or trailers can be written. + /// + /// - Parameter body: A closure that is provided a ``NIOHTTPServer/StreamReset`` instance from which the request + /// stream can be reset with a transport-specific error code. + public consuming func reset(_ body: (consuming NIOHTTPServer.StreamReset) throws -> Void) throws { + self.writerState.markReset(self.streamReset) + return try body(self.streamReset) + } } } diff --git a/Tests/NIOHTTPServerTests/NIOHTTPServerResponseSenderTests.swift b/Tests/NIOHTTPServerTests/NIOHTTPServerResponseSenderTests.swift index e7403ed..0ae01f7 100644 --- a/Tests/NIOHTTPServerTests/NIOHTTPServerResponseSenderTests.swift +++ b/Tests/NIOHTTPServerTests/NIOHTTPServerResponseSenderTests.swift @@ -27,7 +27,11 @@ struct NIOHTTPServerResponseSenderTests { // Sending an informational header with a non-1xx status code shouldn't be allowed try await #require(processExitsWith: .failure) { let (outboundWriter, _) = NIOAsyncChannelOutboundWriter.makeTestingWriter() - var sender = NIOHTTPServer.ResponseSender(writer: outboundWriter, writerState: .init()) + var sender = NIOHTTPServer.ResponseSender( + writer: outboundWriter, + writerState: .init(), + streamReset: .unavailable + ) try await sender.sendInformational(.init(status: .ok, headerFields: [.contentType: "application/json"])) } @@ -37,7 +41,11 @@ struct NIOHTTPServerResponseSenderTests { @available(anyAppleOS 26.0, *) func testSendMultipleInformationalResponses() async throws { let (outboundWriter, sink) = NIOAsyncChannelOutboundWriter.makeTestingWriter() - var sender = NIOHTTPServer.ResponseSender(writer: outboundWriter, writerState: .init()) + var sender = NIOHTTPServer.ResponseSender( + writer: outboundWriter, + writerState: .init(), + streamReset: .unavailable + ) // Send two informational responses let firstInfoHead = HTTPResponse(status: .continue, headerFields: [.contentType: "application/json"]) diff --git a/Tests/NIOHTTPServerTests/NIOHTTPServerWriterTests.swift b/Tests/NIOHTTPServerTests/NIOHTTPServerWriterTests.swift index 6ed44da..bfa6dd1 100644 --- a/Tests/NIOHTTPServerTests/NIOHTTPServerWriterTests.swift +++ b/Tests/NIOHTTPServerTests/NIOHTTPServerWriterTests.swift @@ -31,7 +31,11 @@ struct NIOHTTPServerWriterTests { @available(anyAppleOS 26.0, *) func testSingleWriteAndConclude() async throws { let (writer, sink) = NIOAsyncChannelOutboundWriter.makeTestingWriter() - let responseWriter = NIOHTTPServer.ResponseSender.Writer(writer: writer, writerState: .init()) + let responseWriter = NIOHTTPServer.ResponseSender.Writer( + writer: writer, + writerState: .init(), + streamReset: .unavailable + ) var buffer = UniqueArray(copying: [self.bodySampleOne]) try await responseWriter.finish(buffer: &buffer, finalElement: self.trailerSampleOne) @@ -50,7 +54,11 @@ struct NIOHTTPServerWriterTests { @available(anyAppleOS 26.0, *) func testProduceMultipleElementsAndSingleTrailer() async throws { let (writer, sink) = NIOAsyncChannelOutboundWriter.makeTestingWriter() - var responseWriter = NIOHTTPServer.ResponseSender.Writer(writer: writer, writerState: .init()) + var responseWriter = NIOHTTPServer.ResponseSender.Writer( + writer: writer, + writerState: .init(), + streamReset: .unavailable + ) var buffer = UniqueArray(copying: [self.bodySampleOne]) try await responseWriter.write(buffer: &buffer) @@ -73,7 +81,11 @@ struct NIOHTTPServerWriterTests { @available(anyAppleOS 26.0, *) func testNoBodyJustTrailers() async throws { let (writer, sink) = NIOAsyncChannelOutboundWriter.makeTestingWriter() - let responseWriter = NIOHTTPServer.ResponseSender.Writer(writer: writer, writerState: .init()) + let responseWriter = NIOHTTPServer.ResponseSender.Writer( + writer: writer, + writerState: .init(), + streamReset: .unavailable + ) try await responseWriter.finish(trailer: self.trailerSampleTwo) diff --git a/Tests/NIOHTTPServerTests/Utilities/NegotiatedClientConnection.swift b/Tests/NIOHTTPServerTests/Utilities/NegotiatedClientConnection.swift index 9e438b0..515b2c1 100644 --- a/Tests/NIOHTTPServerTests/Utilities/NegotiatedClientConnection.swift +++ b/Tests/NIOHTTPServerTests/Utilities/NegotiatedClientConnection.swift @@ -60,6 +60,20 @@ enum NegotiatedClientConnection { } } } + + /// Opens a stream *without* the `HTTP2FramePayloadToHTTPClientCodec`, exposing raw + /// `HTTP2Frame.FramePayload`s. Unlike ``openStream()``, this lets a test observe frames the codec would + /// otherwise drop — notably `RST_STREAM`. + func openRawStream() async throws -> NIOAsyncChannel { + try await self.http2StreamMultiplexer.openStream { channel in + channel.eventLoop.makeCompletedFuture { + try NIOAsyncChannel( + wrappingChannelSynchronously: channel, + configuration: .init(isOutboundHalfClosureEnabled: true) + ) + } + } + } } }