Correctly discard only inactive streams. (#11)

Motivation:

Long-lived connections that have many inactive streams would cause
crashes. Crashes are, in my professional opinion, less than ideal
when they occur under expected usage cases.

Modifications:

- Fixed an inverted boolean check that would discard only *active*
  streams.
- Made the cached stream count configurable, removing a TODO.
- Added tests.

Result:

Fewer crashes, better software.

Author: Lukasa

Sources/NIOHTTP2/HTTP2Parser.swift

@@ -156,16 +156,27 @@ public final class HTTP2Parser: ChannelInboundHandler, ChannelOutboundHandler {
     private let mode: ParserMode
     private let initialSettings: [HTTP2Setting]
     private let reentrancyManager = ReentrancyManager()
+    private let maxCachedClosedStreams: Int
 
-    public init(mode: ParserMode, initialSettings: [HTTP2Setting] = nioDefaultSettings) {
+    /// Create a `HTTP2Parser`.
+    ///
+    /// - parameters:
+    ///     - mode: The mode for the parser to operate in: server or client.
+    ///     - initialSettings: The initial settings used for the connection, to be sent in the
+    ///         connection preamble.
+    ///     - maxCachedClosedStreams: The maximum number of streams for which metadata will be preserved
+    ///         to handle delayed frames (e.g. DATA frames that were already in flight after stream reset
+    ///         or GOAWAY).
+    public init(mode: ParserMode, initialSettings: [HTTP2Setting] = nioDefaultSettings, maxCachedClosedStreams: Int = 1024) {
         self.mode = mode
         self.initialSettings = initialSettings
+        self.maxCachedClosedStreams = maxCachedClosedStreams
     }
 
     public func channelActive(ctx: ChannelHandlerContext) {
         self.session = NGHTTP2Session(mode: self.mode,
                                       allocator: ctx.channel.allocator,
-                                      maxCachedStreamIDs: 1024,  // TODO(cory): Make configurable
+                                      maxCachedStreamIDs: self.maxCachedClosedStreams,
                                       frameReceivedHandler: { ctx.fireChannelRead(self.wrapInboundOut($0)) },
                                       sendFunction: { ctx.write(self.wrapOutboundOut($0), promise: $1) },
                                       userEventFunction: { ctx.fireUserInboundEventTriggered($0) })

Sources/NIOHTTP2/NGHTTP2Session.swift

@@ -285,7 +285,7 @@ fileprivate struct StreamManager {
     // Discard old and unnecessary streams.
     private mutating func purgeOldStreams() {
         while self.streamMap.count >= maxSize {
-            let lowestStreamID = self.streamMap.filter { $0.value.active }.keys.sorted().first { $0 != 0 && $0 != Int32.max }!
+            let lowestStreamID = self.streamMap.filter { !$0.value.active }.keys.sorted().first { $0 != 0 && $0 != Int32.max }!
             self.streamMap.removeValue(forKey: lowestStreamID)
         }
     }

Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift

@@ -55,6 +55,8 @@ extension SimpleClientServerTests {
                 ("testStreamClosedViaGoaway", testStreamClosedViaGoaway),
                 ("testStreamCloseEventForRstStreamFiresAfterFrame", testStreamCloseEventForRstStreamFiresAfterFrame),
                 ("testStreamCloseEventForGoawayFiresAfterFrame", testStreamCloseEventForGoawayFiresAfterFrame),
+                ("testManyConcurrentInactiveStreams", testManyConcurrentInactiveStreams),
+                ("testDontRemoveActiveStreams", testDontRemoveActiveStreams),
            ]
    }
 }

Tests/NIOHTTP2Tests/SimpleClientServerTests.swift

@@ -183,9 +183,9 @@ class SimpleClientServerTests: XCTestCase {
     }
 
     /// Establish a basic HTTP/2 connection.
-    func basicHTTP2Connection() throws {
-        XCTAssertNoThrow(try self.clientChannel.pipeline.add(handler: HTTP2Parser(mode: .client)).wait())
-        XCTAssertNoThrow(try self.serverChannel.pipeline.add(handler: HTTP2Parser(mode: .server)).wait())
+    func basicHTTP2Connection(maxCachedClosedStreams: Int = 1024) throws {
+        XCTAssertNoThrow(try self.clientChannel.pipeline.add(handler: HTTP2Parser(mode: .client, maxCachedClosedStreams: maxCachedClosedStreams)).wait())
+        XCTAssertNoThrow(try self.serverChannel.pipeline.add(handler: HTTP2Parser(mode: .server, maxCachedClosedStreams: maxCachedClosedStreams)).wait())
         try self.assertDoHandshake(client: self.clientChannel, server: self.serverChannel)
     }
 
@@ -1230,4 +1230,91 @@ class SimpleClientServerTests: XCTestCase {
         XCTAssertNoThrow(try self.clientChannel.finish())
         XCTAssertNoThrow(try self.serverChannel.finish())
     }
+
+    func testManyConcurrentInactiveStreams() throws {
+        let maxCachedClosedStreams = 128
+
+        // Begin by getting the connection up.
+        try self.basicHTTP2Connection(maxCachedClosedStreams: maxCachedClosedStreams)
+
+        // Obtain some request data.
+        let requestHeaders = HTTPHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+        var requestBody = self.clientChannel.allocator.buffer(capacity: 128)
+        requestBody.write(staticString: "A simple HTTP/2 request.")
+        let responseHeaders = HTTPHeaders([(":status", "200"), ("content-length", "0")])
+
+        // We're going to initiate and then close more than maxCachedClosedStreams streams.
+        // Nothing bad should happen here.
+        for _ in 0...maxCachedClosedStreams {
+            // We're now going to try to send a request from the client to the server.
+            let clientStreamID = HTTP2StreamID()
+            let reqFrame = HTTP2Frame(streamID: clientStreamID, payload: .headers(requestHeaders))
+            var reqBodyFrame = HTTP2Frame(streamID: clientStreamID, payload: .data(.byteBuffer(requestBody)))
+            reqBodyFrame.endStream = true
+
+            let serverStreamID = try self.assertFramesRoundTrip(frames: [reqFrame, reqBodyFrame], sender: self.clientChannel, receiver: self.serverChannel).first!.streamID
+
+            // Let's send a quick response back.
+            var respFrame = HTTP2Frame(streamID: serverStreamID, payload: .headers(responseHeaders))
+            respFrame.endStream = true
+            try self.assertFramesRoundTrip(frames: [respFrame], sender: self.serverChannel, receiver: self.clientChannel)
+        }
+
+        XCTAssertNoThrow(try self.clientChannel.finish())
+        XCTAssertNoThrow(try self.serverChannel.finish())
+    }
+
+    func testDontRemoveActiveStreams() throws {
+        // This is a bit of a regression test, not a generally useful one. See https://github.com/apple/swift-nio-http2/pull/11/
+        // for more.
+        let maxCachedClosedStreams = 128
+
+        // Begin by getting the connection up.
+        try self.basicHTTP2Connection(maxCachedClosedStreams: maxCachedClosedStreams)
+
+        // Obtain some request data.
+        let requestHeaders = HTTPHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+        var requestBody = self.clientChannel.allocator.buffer(capacity: 128)
+        requestBody.write(staticString: "A simple HTTP/2 request.")
+        let responseHeaders = HTTPHeaders([(":status", "200"), ("content-length", "0")])
+
+        // We're going to initiate and then close one fewer than maxCachedClosedStreams streams.
+        // Nothing bad should happen here.
+        for _ in 0..<(maxCachedClosedStreams - 2) {
+            let clientStreamID = HTTP2StreamID()
+            let reqFrame = HTTP2Frame(streamID: clientStreamID, payload: .headers(requestHeaders))
+            var reqBodyFrame = HTTP2Frame(streamID: clientStreamID, payload: .data(.byteBuffer(requestBody)))
+            reqBodyFrame.endStream = true
+
+            let serverStreamID = try self.assertFramesRoundTrip(frames: [reqFrame, reqBodyFrame], sender: self.clientChannel, receiver: self.serverChannel).first!.streamID
+
+            // Let's send a quick response back.
+            var respFrame = HTTP2Frame(streamID: serverStreamID, payload: .headers(responseHeaders))
+            respFrame.endStream = true
+            try self.assertFramesRoundTrip(frames: [respFrame], sender: self.serverChannel, receiver: self.clientChannel)
+        }
+
+        // Ok, now we're going to open *two* streams. In the old, broken code, the opening of the second
+        // stream would discard the first *open* stream, instead of one of the dead old ones.
+        let clientStreamIDs = (0..<2).map { _ in HTTP2StreamID() }
+        let clientFrames = clientStreamIDs.map { HTTP2Frame(streamID: $0, payload: .headers(requestHeaders)) }
+        let serverStreamIDs = try self.assertFramesRoundTrip(frames: clientFrames, sender: self.clientChannel, receiver: self.serverChannel).map { $0.streamID }
+
+        // Now we're going to send the two data frames for these streams.
+        // In the old broken version of this code, this will fail because we accidentally deleted one of our two *open*
+        // stream states, instead of one of the 1024 closed ones. Booleans are hard.
+        let clientDataFrames = clientStreamIDs.map { HTTP2Frame(streamID: $0, payload: .data(.byteBuffer(requestBody))) }
+        try self.assertFramesRoundTrip(frames: clientDataFrames, sender: self.clientChannel, receiver: self.serverChannel)
+
+        // Clean it up with the server now.
+        let serverFrames = serverStreamIDs.map { streamID -> HTTP2Frame in
+            var respFrame = HTTP2Frame(streamID: streamID, payload: .headers(responseHeaders))
+            respFrame.endStream = true
+            return respFrame
+        }
+        try self.assertFramesRoundTrip(frames: serverFrames, sender: self.serverChannel, receiver: self.clientChannel)
+
+        XCTAssertNoThrow(try self.clientChannel.finish())
+        XCTAssertNoThrow(try self.serverChannel.finish())
+    }
 }