Repository: apple/swift-nio-http2 (swift)
Issue: Data frames with END_STREAM that terminate a HTTP/2 stream are not checked for window resizing by HTTP2StreamMultiplexer
When we receive a `DATA` frame with `END_STREAM` set that terminates a stream, we don't emit the correct events to cause the `HTTP2StreamMultiplexer` to emit `WINDOW_UPDATE` frames for the connection. In the worst possible cases this could actually lead to deadlock if we allow the connection window to go to zero here.
Pull request description: Motivation:

Frames that close a stream may consume some of the connection window. We
were not correctly accounting for that in the code, which meant that
frames that carried END_STREAM would not notify the stream multiplexer
that they had consumed connection window space. In the worst cases this
could lead to resource starvation due to consumption of this window
without causing us to emit new WINDOW_UPDATE frames.

Modifications:

StreamClosedEvents should be followed by flow control change events.

Result:

We will correctly update the connection window when a stream is closed.
Resolves #173.

Patch:
diff --git a/Sources/NIOHTTP2/HTTP2ChannelHandler.swift b/Sources/NIOHTTP2/HTTP2ChannelHandler.swift
index 51d8a2a0..898d5cf7 100644
--- a/Sources/NIOHTTP2/HTTP2ChannelHandler.swift
+++ b/Sources/NIOHTTP2/HTTP2ChannelHandler.swift
@@ -505,7 +505,9 @@ extension NIOHTTP2Handler {
 
         switch stateChange {
         case .streamClosed(let streamClosedData):
+            self.outboundBuffer.connectionWindowSize = streamClosedData.localConnectionWindowSize
             self.inboundEventBuffer.pendingUserEvent(StreamClosedEvent(streamID: streamClosedData.streamID, reason: streamClosedData.reason))
+            self.inboundEventBuffer.pendingUserEvent(NIOHTTP2WindowUpdatedEvent(streamID: .rootStream, inboundWindowSize: streamClosedData.remoteConnectionWindowSize, outboundWindowSize: streamClosedData.localConnectionWindowSize))
 
             let failedWrites = self.outboundBuffer.streamClosed(streamClosedData.streamID)
             let error = NIOHTTP2Errors.StreamClosed(streamID: streamClosedData.streamID, errorCode: streamClosedData.reason ?? .cancel)


Test patch:
diff --git a/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift b/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
index 4241eb8e..6a561a36 100644
--- a/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
+++ b/Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift
@@ -69,6 +69,8 @@ extension SimpleClientServerTests {
                 ("testEnforcingMaxHeaderListSize", testEnforcingMaxHeaderListSize),
                 ("testForbidsExceedingMaxHeaderListSizeBeforeDecoding", testForbidsExceedingMaxHeaderListSizeBeforeDecoding),
                 ("testForbidsExceedingMaxHeaderListSizeBeforeDecodingSingleFrame", testForbidsExceedingMaxHeaderListSizeBeforeDecodingSingleFrame),
+                ("testNoStreamWindowUpdateOnEndStreamFrameFromServer", testNoStreamWindowUpdateOnEndStreamFrameFromServer),
+                ("testNoStreamWindowUpdateOnEndStreamFrameFromClient", testNoStreamWindowUpdateOnEndStreamFrameFromClient),
            ]
    }
 }
diff --git a/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift b/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
index 6d139dc1..3f8eec36 100644
--- a/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
+++ b/Tests/NIOHTTP2Tests/SimpleClientServerTests.swift
@@ -207,9 +207,11 @@ class SimpleClientServerTests: XCTestCase {
     }
 
     /// Establish a basic HTTP/2 connection.
-    func basicHTTP2Connection(withMultiplexerCallback multiplexerCallback: ((Channel, HTTP2StreamID) -> EventLoopFuture<Void>)? = nil) throws {
-        XCTAssertNoThrow(try self.clientChannel.pipeline.addHandler(NIOHTTP2Handler(mode: .client)).wait())
-        XCTAssertNoThrow(try self.serverChannel.pipeline.addHandler(NIOHTTP2Handler(mode: .server)).wait())
+    func basicHTTP2Connection(clientSettings: HTTP2Settings = nioDefaultSettings,
+                              serverSettings: HTTP2Settings = nioDefaultSettings,
+                              withMultiplexerCallback multiplexerCallback: ((Channel, HTTP2StreamID) -> EventLoopFuture<Void>)? = nil) throws {
+        XCTAssertNoThrow(try self.clientChannel.pipeline.addHandler(NIOHTTP2Handler(mode: .client, initialSettings: clientSettings)).wait())
+        XCTAssertNoThrow(try self.serverChannel.pipeline.addHandler(NIOHTTP2Handler(mode: .server, initialSettings: serverSettings)).wait())
 
         if let multiplexerCallback = multiplexerCallback {
             XCTAssertNoThrow(try self.clientChannel.pipeline.addHandler(HTTP2StreamMultiplexer(mode: .client,
@@ -220,7 +222,8 @@ class SimpleClientServerTests: XCTestCase {
                                                                                                  inboundStreamStateInitializer: multiplexerCallback)).wait())
         }
 
-        try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel)
+        try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel,
+                                   clientSettings: clientSettings, serverSettings: serverSettings)
     }
 
     /// Establish a basic HTTP/2 connection where the HTTP2Parser handler is added after the channel has been activated.
@@ -808,8 +811,8 @@ class SimpleClientServerTests: XCTestCase {
         try self.assertFramesRoundTrip(frames: [respFrame], sender: self.serverChannel, receiver: self.clientChannel)
 
         // Now the streams are closed, they should have seen user events.
-        XCTAssertEqual(clientHandler.events.count, 4)
-        XCTAssertEqual(serverHandler.events.count, 4)
+        XCTAssertEqual(clientHandler.events.count, 5)
+        XCTAssertEqual(serverHandler.events.count, 5)
         XCTAssertEqual(clientHandler.events[3] as? StreamClosedEvent, StreamClosedEvent(streamID: clientStreamID, reason: nil))
         XCTAssertEqual(serverHandler.events[3] as? StreamClosedEvent, StreamClosedEvent(streamID: serverStreamID, reason: nil))
 
@@ -845,8 +848,8 @@ class SimpleClientServerTests: XCTestCase {
         try self.assertFramesRoundTrip(frames: [rstStreamFrame], sender: self.serverChannel, receiver: self.clientChannel)
 
         // Now the streams are closed, they should have seen user events.
-        XCTAssertEqual(clientHandler.events.count, 2)
-        XCTAssertEqual(serverHandler.events.count, 2)
+        XCTAssertEqual(clientHandler.events.count, 3)
+        XCTAssertEqual(serverHandler.events.count, 3)
         XCTAssertEqual(clientHandler.events[1] as? StreamClosedEvent, StreamClosedEvent(streamID: clientStreamID, reason: .cancel))
         XCTAssertEqual(serverHandler.events[1] as? StreamClosedEvent, StreamClosedEvent(streamID: serverStreamID, reason: .cancel))
 
@@ -1782,4 +1785,107 @@ class SimpleClientServerTests: XCTestCase {
 
         try self.clientChannel.assertReceivedFrame().assertGoAwayFrame(lastStreamID: .maxID, errorCode: UInt32(HTTP2ErrorCode.protocolError.networkCode), opaqueData: nil)
     }
+
+    func testNoStreamWindowUpdateOnEndStreamFrameFromServer() throws {
+        let maxFrameSize = (1 << 24) - 1  // Max value of SETTINGS_MAX_FRAME_SIZE is 2^24-1.
+        let clientSettings = nioDefaultSettings + [HTTP2Setting(parameter: .maxFrameSize, value: maxFrameSize)]
+
+        // Begin by getting the connection up.
+        // We add the stream multiplexer here because it manages automatic flow control, and because we want the server
+        // to correctly confirm we don't receive a frame for the child stream.
+        let childHandler = FrameRecorderHandler()
+        try self.basicHTTP2Connection(clientSettings: clientSettings) { channel, streamID in
+            // Here we send a large response: 65535 bytes in size.
+            let responseHeaders = HPACKHeaders([(":status", "200"), ("content-length", "65535")])
+
+            var responseBody = self.clientChannel.allocator.buffer(capacity: 65535)
+            responseBody.writeBytes(Array(repeating: UInt8(0x04), count: 65535))
+
+            let respFrame = HTTP2Frame(streamID: streamID, payload: .headers(.init(headers: responseHeaders)))
+            channel.write(respFrame, promise: nil)
+
+            // Now prepare the large body.
+            let respBodyFrame = HTTP2Frame(streamID: streamID, payload: .data(.init(data: .byteBuffer(responseBody), endStream: true)))
+            channel.writeAndFlush(respBodyFrame, promise: nil)
+
+            return channel.pipeline.addHandler(childHandler)
+        }
+
+        // Now we're going to send a small request.
+        let headers = HPACKHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+
+        // We're going to open a stream and queue up the frames for that stream.
+        let handler = try self.clientChannel.pipeline.handler(type: HTTP2StreamMultiplexer.self).wait()
+        var reqFrame: HTTP2Frame? = nil
+
+        handler.createStreamChannel(promise: nil) { channel, streamID in
+            // We need END_STREAM set here, because that will force the stream to be closed on the response.
+            reqFrame = HTTP2Frame(streamID: streamID, payload: .headers(.init(headers: headers, endStream: true)))
+            channel.writeAndFlush(reqFrame, promise: nil)
+            return channel.eventLoop.makeSucceededFuture(())
+        }
+        self.clientChannel.embeddedEventLoop.run()
+
+        // Ok, we now want to send this data to the server.
+        self.interactInMemory(self.clientChannel, self.serverChannel)
+
+        // The server should have seen 1 window update frame for the connection only.
+        try self.serverChannel.assertReceivedFrame().assertWindowUpdateFrame(streamID: 0, windowIncrement: 65535)
+
+        // And only the request frame frame for the child stream, as there was no need to open its stream window.
+        childHandler.receivedFrames.assertFramesMatch([reqFrame!])
+
+        // No other frames should be emitted, though the client may have many in a child stream.
+        self.serverChannel.assertNoFramesReceived()
+        XCTAssertNoThrow(XCTAssertTrue(try self.clientChannel.finish().isClean))
+        XCTAssertNoThrow(XCTAssertTrue(try self.serverChannel.finish().isClean))
+    }
+
+    func testNoStreamWindowUpdateOnEndStreamFrameFromClient() throws {
+        let maxFrameSize = (1 << 24) - 1  // Max value of SETTINGS_MAX_FRAME_SIZE is 2^24-1.
+        let serverSettings = nioDefaultSettings + [HTTP2Setting(parameter: .maxFrameSize, value: maxFrameSize)]
+
+        // Begin by getting the connection up.
+        // We add the stream multiplexer here because it manages automatic flow control, and because we want the server
+        // to correctly confirm we don't receive a frame for the child stream.
+        let childHandler = FrameRecorderHandler()
+        try self.basicHTTP2Connection(serverSettings: serverSettings) { channel, streamID in
+            return channel.eventLoop.makeSucceededFuture(())
+        }
+
+        // Now we're going to send a large request: 65535 bytes in size
+        let headers = HPACKHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+
+        // We're going to open a stream and queue up the frames for that stream.
+        let handler = try self.clientChannel.pipeline.handler(type: HTTP2StreamMultiplexer.self).wait()
+
+        handler.createStreamChannel(promise: nil) { channel, streamID in
+            let reqFrame = HTTP2Frame(streamID: streamID, payload: .headers(.init(headers: headers)))
+            channel.write(reqFrame, promise: nil)
+
+            var requestBody = self.clientChannel.allocator.buffer(capacity: 65535)
+            requestBody.writeBytes(Array(repeating: UInt8(0x04), count: 65535))
+
+            // Now prepare the large body. We need END_STREAM set.
+            let reqBodyFrame = HTTP2Frame(streamID: streamID, payload: .data(.init(data: .byteBuffer(requestBody), endStream: true)))
+            channel.writeAndFlush(reqBodyFrame, promise: nil)
+
+            return channel.pipeline.addHandler(childHandler)
+        }
+        self.clientChannel.embeddedEventLoop.run()
+
+        // Ok, we now want to send this data to the server.
+        self.interactInMemory(self.clientChannel, self.serverChannel)
+
+        // The client should have seen a window update on the connection.
+        try self.clientChannel.assertReceivedFrame().assertWindowUpdateFrame(streamID: 0, windowIncrement: 65535)
+
+        // And nothing for the child stream, as there was no need to open its stream window.
+        childHandler.receivedFrames.assertFramesMatch([])
+
+        // No other frames should be emitted, though the server may have many in a child stream.
+        self.clientChannel.assertNoFramesReceived()
+        XCTAssertNoThrow(XCTAssertTrue(try self.clientChannel.finish().isClean))
+        XCTAssertNoThrow(XCTAssertTrue(try self.serverChannel.finish().isClean))
+    }
 }