Add reset stream API

Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift

@@ -215,6 +215,7 @@ extension NIOHTTPServer {
                         request: httpRequest,
                         iterator: iterator,
                         outbound: outbound,
+                        streamReset: .unavailable,
                         handler: handler,
                         context: context
                     )

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
+                    )
                 }
             }
         }

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<Handler: HTTPServerRequestHandler>(
         channel: NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>,
         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.

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

Sources/NIOHTTPServer/NIOHTTPServer.swift

@@ -332,6 +332,7 @@ public struct NIOHTTPServer: HTTPServer {
         request: HTTPRequest,
         iterator: consuming sending NIOAsyncChannelInboundStream<HTTPRequestPart>.AsyncIterator,
         outbound: NIOAsyncChannelOutboundWriter<HTTPResponsePart>,
+        streamReset: consuming sending NIOHTTPServer.StreamReset,
         handler: Handler,
         context: ConnectionContext
     ) async throws -> NIOAsyncChannelInboundStream<HTTPRequestPart>.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)")

Sources/NIOHTTPServer/NIOHTTPServerResponseSender.swift

@@ -21,6 +21,7 @@ extension NIOHTTPServer {
     public struct ResponseSender: HTTPResponseSender, ~Copyable {
         let writer: NIOAsyncChannelOutboundWriter<HTTPResponsePart>
         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<Wrapped> = .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<UInt8> & ~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)
+        }
     }
 }
 

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<HTTPResponsePart>.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<HTTPResponsePart>.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"])

Tests/NIOHTTPServerTests/NIOHTTPServerWriterTests.swift

@@ -31,7 +31,11 @@ struct NIOHTTPServerWriterTests {
     @available(anyAppleOS 26.0, *)
     func testSingleWriteAndConclude() async throws {
         let (writer, sink) = NIOAsyncChannelOutboundWriter<HTTPResponsePart>.makeTestingWriter()
-        let responseWriter = NIOHTTPServer.ResponseSender.Writer(writer: writer, writerState: .init())
+        let responseWriter = NIOHTTPServer.ResponseSender.Writer(
+            writer: writer,
+            writerState: .init(),
+            streamReset: .unavailable
+        )
 
         var buffer = UniqueArray<UInt8>(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<HTTPResponsePart>.makeTestingWriter()
-        var responseWriter = NIOHTTPServer.ResponseSender.Writer(writer: writer, writerState: .init())
+        var responseWriter = NIOHTTPServer.ResponseSender.Writer(
+            writer: writer,
+            writerState: .init(),
+            streamReset: .unavailable
+        )
 
         var buffer = UniqueArray<UInt8>(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<HTTPResponsePart>.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)