Better handle new streams when server is quiescing

Sources/NIOHTTP2/ConnectionStateMachine/ConnectionStreamsState.swift

@@ -181,14 +181,16 @@ struct ConnectionStreamState {
         streamID: HTTP2StreamID,
         ignoreRecentlyReset: Bool,
         ignoreClosed: Bool = false,
+        isQuiescing: Bool = false,
         _ modifier: (inout HTTP2StreamStateMachine) -> StateMachineResultWithStreamEffect
     ) -> StateMachineResultWithStreamEffect {
         guard let result = self.activeStreams.autoClosingTransform(streamID: streamID, modifier) else {
             return StateMachineResultWithStreamEffect(
                 result: self.streamMissing(
                     streamID: streamID,
                     ignoreRecentlyReset: ignoreRecentlyReset,
-                    ignoreClosed: ignoreClosed
+                    ignoreClosed: ignoreClosed,
+                    isQuiescing: isQuiescing
                 ),
                 effect: nil
             )
@@ -218,11 +220,28 @@ struct ConnectionStreamState {
         _ modifier: (inout HTTP2StreamStateMachine) -> StateMachineResultWithStreamEffect
     ) -> StateMachineResultWithStreamEffect {
         guard let result = self.activeStreams.autoClosingTransform(streamID: streamID, modifier) else {
-            // We never ignore recently reset streams here, as this should only ever be used when *sending* frames.
-            return StateMachineResultWithStreamEffect(
-                result: self.streamMissing(streamID: streamID, ignoreRecentlyReset: false, ignoreClosed: false),
-                effect: nil
-            )
+            // This state can be reached when a RST_STREAM frame is sent by this peer but the stream
+            // doesn't exist. This can happen for a few reasons, including:
+            //
+            // 1. The RST_STREAM frame is being sent because the stream wasn't accepted (i.e. the
+            //    client opening the stream raced with the server sending a GOAWAY frame). In this
+            //    case the state machine doesn't know about the stream but does need to send
+            //    a RST_STREAM frame.
+            // 2. An implementation or user error where the stream genuinely doesn't exist and
+            //    attempting to send a RST_STREAM frame is an error.
+            if streamID.isClientInitiated, streamID > self.lastClientStreamID {
+                return StateMachineResultWithStreamEffect(result: .succeed, effect: nil)
+            } else {
+                return StateMachineResultWithStreamEffect(
+                    result: self.streamMissing(
+                        streamID: streamID,
+                        ignoreRecentlyReset: false,
+                        ignoreClosed: false,
+                        isQuiescing: true
+                    ),
+                    effect: nil
+                )
+            }
         }
 
         guard let effect = result.effect, effect.closedStream else {
@@ -324,7 +343,8 @@ struct ConnectionStreamState {
     private func streamMissing(
         streamID: HTTP2StreamID,
         ignoreRecentlyReset: Bool,
-        ignoreClosed: Bool
+        ignoreClosed: Bool,
+        isQuiescing: Bool
     ) -> StateMachineResult {
         if ignoreRecentlyReset && self.recentlyResetStreams.contains(streamID) {
             return .ignoreFrame
@@ -333,11 +353,20 @@ struct ConnectionStreamState {
         switch streamID.mayBeInitiatedBy(.client) {
         case true where streamID > self.lastClientStreamID,
             false where streamID > self.lastServerStreamID:
-            // The stream in question is idle.
-            return .connectionError(
-                underlyingError: NIOHTTP2Errors.noSuchStream(streamID: streamID),
-                type: .protocolError
-            )
+            if isQuiescing {
+                // This peer quiescing and the remote peer opening a stream raced. Reject the stream.
+                return .streamError(
+                    streamID: streamID,
+                    underlyingError: NIOHTTP2Errors.streamError(streamID: streamID, baseError: NIOHTTP2Errors.createdStreamAfterGoaway()),
+                    type: .refusedStream
+                )
+            } else {
+                // The stream in question is idle.
+                return .connectionError(
+                    underlyingError: NIOHTTP2Errors.noSuchStream(streamID: streamID),
+                    type: .protocolError
+                )
+            }
         default:
             // This stream must have already been closed.
             if ignoreClosed {

Sources/NIOHTTP2/ConnectionStateMachine/FrameReceivingStates/ReceivingHeadersState.swift

@@ -115,7 +115,7 @@ extension ReceivingHeadersState where Self: LocallyQuiescingState {
         }
 
         // At this stage we've quiesced, so the remote peer is not allowed to create new streams.
-        let result = self.streamState.modifyStreamState(streamID: streamID, ignoreRecentlyReset: true) {
+        let result = self.streamState.modifyStreamState(streamID: streamID, ignoreRecentlyReset: true, isQuiescing: true) {
             $0.receiveHeaders(
                 headers: headers,
                 validateHeaderBlock: validateHeaderBlock,

Tests/NIOHTTP2Tests/ConnectionStateMachineTests.swift

@@ -1266,6 +1266,60 @@ class ConnectionStateMachineTests: XCTestCase {
         assertSucceeds(temporaryClient.receiveRstStream(streamID: streamOne, reason: .refusedStream))
     }
 
+    func testClientInitiatesNewStreamBeforeReceivingAlreadySentGoawayWhenServerLocallyQuiesced() {
+        // Tests the behaviour of the server when it has open streams, sends a GOAWAY frame (so is
+        // in locally quiescing state), and then receives HEADERS for a stream whose ID is less than
+        // the last stream ID sent in the GOAWAY frame.
+        //
+        // The expected behaviour is that the server should refuse the stream (by sending RST_STREAM
+        // frame) and emit a stream error.
+        let streamOne = HTTP2StreamID(1)
+        let streamThree = HTTP2StreamID(3)
+
+        self.exchangePreamble()
+
+        // Open stream one.
+        assertSucceeds(
+            self.client.sendHeaders(
+                streamID: streamOne,
+                headers: ConnectionStateMachineTests.requestHeaders,
+                isEndStreamSet: false
+            )
+        )
+        assertSucceeds(
+            self.server.receiveHeaders(
+                streamID: streamOne,
+                headers: ConnectionStateMachineTests.requestHeaders,
+                isEndStreamSet: false
+            )
+        )
+
+        // Server begins quiescing. It enters the locally quiesced state.
+        assertSucceeds(self.server.sendGoaway(lastStreamID: .maxID))
+
+        // Client opens stream three (important: it hasn't received the GOAWAY frame).
+        assertSucceeds(
+            self.client.sendHeaders(
+                streamID: streamThree,
+                headers: ConnectionStateMachineTests.requestHeaders,
+                isEndStreamSet: false
+            )
+        )
+        // Server receives headers for stream three. It should throw a stream error and as a
+        // result, respond with a RST_STREAM frame.
+        assertStreamError(
+            type: .refusedStream,
+            self.server.receiveHeaders(
+                streamID: streamThree,
+                headers: ConnectionStateMachineTests.requestHeaders,
+                isEndStreamSet: false
+            )
+        )
+
+        assertSucceeds(self.server.sendRstStream(streamID: streamOne, reason: .refusedStream))
+        assertSucceeds(self.client.receiveRstStream(streamID: streamOne, reason: .refusedStream))
+    }
+
     func testHeadersOnClosedStreamAfterClientGoaway() {
         let streamOne = HTTP2StreamID(1)
         let streamTwo = HTTP2StreamID(2)

Tests/NIOHTTP2Tests/SimpleClientServerInlineStreamMultiplexerTests.swift

@@ -427,6 +427,70 @@ class SimpleClientServerInlineStreamMultiplexerTests: XCTestCase {
         XCTAssertNoThrow(try self.serverChannel.finish())
     }
 
+    func testOpenStreamBeforeReceivingGoAwayWhenServerLocallyQuiesced() throws {
+        let serverFrameRecorder = InboundFramePayloadRecorder()
+        try self.basicHTTP2Connection { channel in
+            channel.pipeline.addHandler(serverFrameRecorder)
+        }
+
+        let http2Handler = self.clientChannel.pipeline.handler(type: NIOHTTP2Handler.self)
+        let multiplexer = try http2Handler.flatMap { $0.multiplexer }.wait()
+
+        let clientFrameRecorder = InboundFramePayloadRecorder()
+        let streamOneFuture = multiplexer.createStreamChannel { channel in
+            channel.eventLoop.makeCompletedFuture {
+                try channel.pipeline.syncOperations.addHandler(clientFrameRecorder)
+            }
+        }
+        self.clientChannel.embeddedEventLoop.run()
+        let streamOne = try streamOneFuture.wait()
+
+        let headers: HPACKHeaders = [
+            ":path": "/",
+            ":method": "GET",
+            ":scheme": "http",
+            ":authority": "localhost",
+        ]
+        try streamOne.writeAndFlush(HTTP2Frame.FramePayload.headers(.init(headers: headers))).wait()
+        self.interactInMemory(self.clientChannel, self.serverChannel)
+        serverFrameRecorder.receivedFrames.assertFramePayloadsMatch([.headers(.init(headers: headers))])
+
+        // Create stream two, but don't write on it (yet).
+        let streamTwoFuture = multiplexer.createStreamChannel { channel in
+            channel.eventLoop.makeCompletedFuture {
+                try channel.pipeline.syncOperations.addHandler(clientFrameRecorder)
+            }
+        }
+        self.clientChannel.embeddedEventLoop.run()
+        let streamTwo = try streamTwoFuture.wait()
+
+        // Send the GOAWAY from the server.
+        let goAway = HTTP2Frame(
+            streamID: .rootStream,
+            payload: .goAway(lastStreamID: .maxID, errorCode: .noError, opaqueData: nil)
+        )
+        try self.serverChannel.writeAndFlush(goAway).wait()
+        // Send HEADERS on the second client stream.
+        try streamTwo.writeAndFlush(HTTP2Frame.FramePayload.headers(.init(headers: headers))).wait()
+
+        // Interacting in memory to exchange frames.
+        self.interactInMemory(self.clientChannel, self.serverChannel, expectError: true) { error in
+            if let error = error as? NIOHTTP2Errors.StreamError {
+                XCTAssert(error.baseError is NIOHTTP2Errors.CreatedStreamAfterGoaway)
+            } else {
+                XCTFail("Expected error to be of type StreamError, got error of type \(type(of: error)).")
+            }
+        }
+
+        // Client receives GOAWAY and RST_STREAM frames.
+        try self.clientChannel.assertReceivedFrame().assertGoAwayFrame(
+            lastStreamID: .maxID,
+            errorCode: 0,
+            opaqueData: nil
+        )
+        clientFrameRecorder.receivedFrames.assertFramePayloadsMatch([.rstStream(.refusedStream)])
+    }
+
     func testSuccessfullyReceiveAndSendPingEvenWhenConnectionIsFullyQuiesced() throws {
         let serverHandler = InboundFramePayloadRecorder()
         try self.basicHTTP2Connection { channel in