Don't fail closeFuture when an error occurs on closing (#487)

Strictly speaking, a `close` operation can't really fail (except when the closing mode isn't supported or the channel is already closed). Even if an error is encountered while closing a channel, or if an error caused the channel to be closed, the end result is the same: the channel will be closed. In this case in particular, when streams are closed from the client (i.e. by sending a RST_STREAM frame) the `HTTP2StreamChannel` would fail the channel's `closeFuture`. This isn't appropriate however, as the channel is successfully closed. This change stops failing the stream channel's `closeFuture`. Instead, if the close happens uncleanly, the error will be fired down the pipeline and the `close` method's promise will be failed.
apple · Dec 16, 2024 · 170f4ca · 170f4ca
1 parent 2083f5d
commit 170f4ca
Show file tree

Hide file tree

Showing 14 changed files with 268 additions and 169 deletions.
diff --git a/IntegrationTests/tests_01_allocation_counters/Thresholds/5.10.json b/IntegrationTests/tests_01_allocation_counters/Thresholds/5.10.json
@@ -18,6 +18,6 @@
     "get_100000_headers_canonical_form_trimming_whitespace_from_long_string": 300050,
     "get_100000_headers_canonical_form_trimming_whitespace_from_short_string": 200050,
     "hpack_decoding": 5050,
-    "stream_teardown_100_concurrent": 262550,
-    "stream_teardown_100_concurrent_inline": 261650
+    "stream_teardown_100_concurrent": 252400,
+    "stream_teardown_100_concurrent_inline": 252400
 }
diff --git a/IntegrationTests/tests_01_allocation_counters/Thresholds/5.9.json b/IntegrationTests/tests_01_allocation_counters/Thresholds/5.9.json
@@ -18,6 +18,6 @@
     "get_100000_headers_canonical_form_trimming_whitespace_from_long_string": 300050,
     "get_100000_headers_canonical_form_trimming_whitespace_from_short_string": 200050,
     "hpack_decoding": 5050,
-    "stream_teardown_100_concurrent": 262550,
-    "stream_teardown_100_concurrent_inline": 261650
+    "stream_teardown_100_concurrent": 252400,
+    "stream_teardown_100_concurrent_inline": 252400
 }
diff --git a/IntegrationTests/tests_01_allocation_counters/Thresholds/6.0.json b/IntegrationTests/tests_01_allocation_counters/Thresholds/6.0.json
@@ -18,6 +18,6 @@
     "get_100000_headers_canonical_form_trimming_whitespace_from_long_string": 300050,
     "get_100000_headers_canonical_form_trimming_whitespace_from_short_string": 200050,
     "hpack_decoding": 5050,
-    "stream_teardown_100_concurrent": 262450,
-    "stream_teardown_100_concurrent_inline": 261750
+    "stream_teardown_100_concurrent": 252450,
+    "stream_teardown_100_concurrent_inline": 251550
 }
diff --git a/IntegrationTests/tests_01_allocation_counters/Thresholds/nightly-6.0.json b/IntegrationTests/tests_01_allocation_counters/Thresholds/nightly-6.0.json
@@ -18,6 +18,6 @@
     "get_100000_headers_canonical_form_trimming_whitespace_from_long_string": 300050,
     "get_100000_headers_canonical_form_trimming_whitespace_from_short_string": 200050,
     "hpack_decoding": 5050,
-    "stream_teardown_100_concurrent": 262450,
-    "stream_teardown_100_concurrent_inline": 261750
+    "stream_teardown_100_concurrent": 252450,
+    "stream_teardown_100_concurrent_inline": 251550
 }
diff --git a/IntegrationTests/tests_01_allocation_counters/Thresholds/nightly-main.json b/IntegrationTests/tests_01_allocation_counters/Thresholds/nightly-main.json
@@ -18,6 +18,6 @@
     "get_100000_headers_canonical_form_trimming_whitespace_from_long_string": 300050,
     "get_100000_headers_canonical_form_trimming_whitespace_from_short_string": 200050,
     "hpack_decoding": 5050,
-    "stream_teardown_100_concurrent": 262450,
-    "stream_teardown_100_concurrent_inline": 261750
+    "stream_teardown_100_concurrent": 252450,
+    "stream_teardown_100_concurrent_inline": 251550
 }
diff --git a/Sources/NIOHTTP2/HTTP2StreamChannel.swift b/Sources/NIOHTTP2/HTTP2StreamChannel.swift
@@ -676,7 +676,7 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {
 
         self.eventLoop.execute {
             self.removeHandlers(pipeline: self.pipeline)
-            self.closePromise.succeed(())
+            self.closePromise.succeed()
             if let streamID = self.streamID {
                 self.multiplexer.streamClosed(id: streamID)
             } else {
@@ -701,7 +701,7 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {
 
         self.eventLoop.execute {
             self.removeHandlers(pipeline: self.pipeline)
-            self.closePromise.fail(error)
+            self.closePromise.succeed()
             if let streamID = self.streamID {
                 self.multiplexer.streamClosed(id: streamID)
             } else {

diff --git a/Tests/NIOHTTP2Tests/ConfiguringPipelineInlineMultiplexerTests.swift b/Tests/NIOHTTP2Tests/ConfiguringPipelineInlineMultiplexerTests.swift
@@ -71,7 +71,10 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
             )
         )
 
-        clientMultiplexer.createStreamChannel(promise: nil) { channel in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientMultiplexer.createStreamChannel(promise: streamChannelPromise) { channel in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             channel.writeAndFlush(reqFrame.payload).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
         }
@@ -82,9 +85,12 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
 
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // We should have received a HEADERS and a RST_STREAM frame.
         // The RST_STREAM frame is from closing an incomplete stream on the client side.
@@ -133,7 +139,10 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
             )
         )
 
-        clientMultiplexer.createStreamChannel(promise: nil) { channel in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientMultiplexer.createStreamChannel(promise: streamChannelPromise) { channel in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             channel.writeAndFlush(reqFrame.payload).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
         }
@@ -144,9 +153,12 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
 
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // We should have received a HEADERS and a RST_STREAM frame.
         // The RST_STREAM frame is from closing an incomplete stream on the client side.
@@ -466,7 +478,10 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
             )
         )
 
-        clientMultiplexer.createStreamChannel(promise: nil) { channel in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientMultiplexer.createStreamChannel(promise: streamChannelPromise) { channel in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             channel.writeAndFlush(reqFrame.payload).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
         }
@@ -476,9 +491,13 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
+
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // Assert that the user-provided handler received the
         // HTTP1 parts corresponding to the H2 message sent
@@ -556,7 +575,10 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
             )
         )
 
-        clientMultiplexer.createStreamChannel(promise: nil) { channel in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientMultiplexer.createStreamChannel(promise: streamChannelPromise) { channel in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             channel.writeAndFlush(reqFrame.payload).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
         }
@@ -566,9 +588,13 @@ class ConfiguringPipelineInlineMultiplexerTests: XCTestCase {
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
+
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // Assert that the user-provided handler received the
         // HTTP1 parts corresponding to the H2 message sent

diff --git a/Tests/NIOHTTP2Tests/ConfiguringPipelineTests.swift b/Tests/NIOHTTP2Tests/ConfiguringPipelineTests.swift
@@ -61,7 +61,10 @@ class ConfiguringPipelineTests: XCTestCase {
             )
         )
 
-        clientHandler.createStreamChannel(promise: nil) { channel, streamID in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientHandler.createStreamChannel(promise: streamChannelPromise) { channel, streamID in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             XCTAssertEqual(streamID, HTTP2StreamID(1))
             channel.writeAndFlush(reqFrame).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
@@ -73,9 +76,12 @@ class ConfiguringPipelineTests: XCTestCase {
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
 
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // We should have received a HEADERS and a RST_STREAM frame.
         // The RST_STREAM frame is from closing an incomplete stream on the client side.
@@ -116,7 +122,10 @@ class ConfiguringPipelineTests: XCTestCase {
             )
         )
 
-        clientHandler.createStreamChannel(promise: nil) { channel, streamID in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientHandler.createStreamChannel(promise: streamChannelPromise) { channel, streamID in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             XCTAssertEqual(streamID, HTTP2StreamID(1))
             channel.writeAndFlush(reqFrame).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
@@ -128,9 +137,12 @@ class ConfiguringPipelineTests: XCTestCase {
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
 
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         // We should have received a HEADERS and a RST_STREAM frame.
         // The RST_STREAM frame is from closing an incomplete stream on the client side.
@@ -403,7 +415,10 @@ class ConfiguringPipelineTests: XCTestCase {
             )
         )
 
-        clientHandler.createStreamChannel(promise: nil) { channel, streamID in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientHandler.createStreamChannel(promise: streamChannelPromise) { channel, streamID in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             XCTAssertEqual(streamID, HTTP2StreamID(1))
             channel.writeAndFlush(reqFrame).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
@@ -414,9 +429,13 @@ class ConfiguringPipelineTests: XCTestCase {
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
+
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         let serverChildChannel = try serverChildChannelPromise.futureResult.wait()
         try serverChildChannel.pipeline.handler(type: HTTP1ServerRequestRecorderHandler.self).map { serverRecorder in
@@ -492,7 +511,10 @@ class ConfiguringPipelineTests: XCTestCase {
             )
         )
 
-        clientHandler.createStreamChannel(promise: nil) { channel, streamID in
+        let errorHandler = ErrorEncounteredHandler()
+        let streamChannelPromise = self.clientChannel.eventLoop.makePromise(of: Channel.self)
+        clientHandler.createStreamChannel(promise: streamChannelPromise) { channel, streamID in
+            try? channel.pipeline.syncOperations.addHandler(errorHandler)
             XCTAssertEqual(streamID, HTTP2StreamID(1))
             channel.writeAndFlush(reqFrame).whenComplete { _ in channel.close(promise: requestPromise) }
             return channel.eventLoop.makeSucceededFuture(())
@@ -503,9 +525,13 @@ class ConfiguringPipelineTests: XCTestCase {
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
         self.interactInMemory(self.clientChannel, self.serverChannel)
         (self.clientChannel.eventLoop as! EmbeddedEventLoop).run()
+
+        let streamChannel = try XCTUnwrap(streamChannelPromise.futureResult.wait())
+        XCTAssertNoThrow(try streamChannel.closeFuture.wait())
         XCTAssertThrowsError(try requestPromise.futureResult.wait()) { error in
             XCTAssertTrue(error is NIOHTTP2Errors.StreamClosed)
         }
+        XCTAssertTrue(errorHandler.encounteredError is NIOHTTP2Errors.StreamClosed)
 
         let serverChildChannel = try serverChildChannelPromise.futureResult.wait()
         try serverChildChannel.pipeline.handler(type: HTTP1ServerRequestRecorderHandler.self).map { serverRecorder in