This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new 83eb9486d fix: release buffered data accounting when an incoming
HTTP/2 stream is shut down (#1281)
83eb9486d is described below
commit 83eb9486df341dfaf22ee8d9a37148c1785cb330
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 09:51:03 2026 +0100
fix: release buffered data accounting when an incoming HTTP/2 stream is
shut down (#1281)
Motivation:
`IncomingStreamBuffer.shutdown()` fails the entity outlet and drops
whatever is still buffered, but unlike the other discard paths
(`onRstStreamFrame`, `onDownstreamFinish`) it never subtracted those
bytes from `totalBufferedData`.
One of its callers leaves the connection running: when a peer sends more
data than the stream-level window allows, `onDataFrame` resets that
single stream with FLOW_CONTROL_ERROR and carries on. The bytes still
buffered for the stream, plus the payload of the frame that is dropped
without ever being buffered, keep counting as buffered for the lifetime
of the connection. The connection-level flow controller only emits a
WINDOW_UPDATE while `outstanding + buffered` stays below half of
`incoming-connection-level-buffer-size`, so the leaked accounting
permanently reduces - and once it reaches half the buffer size, stops -
the replenishment of the connection window, and every stream on that
connection stalls.
Modification:
Release the buffer in `IncomingStreamBuffer.shutdown()` via the existing
`discardBuffer()`, and subtract the payload of the frame that trips the
stream-level window check, since that frame is never buffered either.
Result:
Resetting a stream for a flow-control violation releases the
connection-level window its data reserved, so the connection keeps being
replenished and no longer stalls.
Tests:
- New test "release connection-level flow control accounting when a
stream-level window is exceeded" makes a peer overrun the stream window with
the connection-level buffer sized so the discarded frame is more than half of
it, then asserts the connection window is handed back in full. Without the fix
no WINDOW_UPDATE is emitted at all.
- sbt "http2-tests/testOnly
org.apache.pekko.http.impl.engine.http2.Http2ServerSpec" - Not run -
environment failure: `http2-tests / update` cannot resolve
`io.github.summerwind:h2spec_darwin_amd64:2.6.0` ("h2spec_darwin_amd64.tar.gz
not found under
https://github.com/summerwind/h2spec/releases/download/v2.6.0/";), which blocks
the whole module. Verified on CI instead.
- The first CI run failed on the new test only, with "requirement failed:
incoming-connection-level-buffer-size must be > 0": the buffer size was a `val`
in the anonymous setup class, which is initialised after the superclass
constructor has already read `settings`. The constants now live outside that
class.
References:
None - releases buffered-data accounting when the incoming side of a stream
is shut down
---
.../impl/engine/http2/Http2StreamHandling.scala | 13 ++++++++++-
.../http/impl/engine/http2/Http2ServerSpec.scala | 25 ++++++++++++++++++++++
2 files changed, 37 insertions(+), 1 deletion(-)
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2StreamHandling.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2StreamHandling.scala
index 364e523fc..cd8b89b33 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2StreamHandling.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2StreamHandling.scala
@@ -667,6 +667,8 @@ private[http2] trait Http2StreamHandling extends
GraphStageLogic with LogHelper
outstandingStreamWindow -= data.sizeInWindow
if (outstandingStreamWindow < 0) {
+ // the frame is never buffered, so stop reserving connection-level
window for it as well
+ totalBufferedData -= data.payload.length
shutdown()
multiplexer.pushControlFrame(RstStreamFrame(streamId,
ErrorCode.FLOW_CONTROL_ERROR))
// also close response delivery if that has already started
@@ -740,8 +742,17 @@ private[http2] trait Http2StreamHandling extends
GraphStageLogic with LogHelper
s"remaining connection window space now
$outstandingConnectionLevelWindow, total buffered: $totalBufferedData")
}
- def shutdown(): Unit =
+ /**
+ * Tears the incoming side of the stream down. Everything still buffered
is dropped here, so it has to be released
+ * from the connection-level accounting like on the other discard paths:
the stream-level flow control error in
+ * `onDataFrame` resets a single stream and leaves the connection running,
so anything kept counted there would
+ * stall the whole connection for good.
+ */
+ def shutdown(): Unit = {
+ discardBuffer()
+ trailingHeaders = None
if (!outlet.isClosed)
outlet.fail(Http2StreamHandling.ConnectionWasAbortedException)
+ }
}
trait OutStream {
diff --git
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerSpec.scala
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerSpec.scala
index ffd415924..662078d44 100644
---
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerSpec.scala
+++
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerSpec.scala
@@ -763,6 +763,31 @@ class Http2ServerSpec extends Http2SpecWithMaterializer("""
network.sendFrame(DataFrame(TheStreamId, endStream = false,
ByteString("0" * 512001))) // more than default
`incoming-stream-level-buffer-size = 512kB`
network.expectRST_STREAM(TheStreamId, ErrorCode.FLOW_CONTROL_ERROR)
})
+ // The connection-level window is only replenished while `outstanding
+ buffered` stays below half of the
+ // buffer size, so a buffer of a bit more than twice the discarded
frame makes the release observable: the
+ // WINDOW_UPDATE below can only arrive when the frame the server
dropped stopped counting as buffered.
+ // These have to live outside of the setup below because `settings` is
read while it is being constructed.
+ val OversizedFrameSize = 512001 // more than default
`incoming-stream-level-buffer-size = 512kB`
+ val ConnectionBufferSize = 700000
+ "release connection-level flow control accounting when a stream-level
window is exceeded"
+ .inAssertAllStagesStopped(new WaitingForRequestData {
+ override def settings: ServerSettings =
+
super.settings.mapHttp2Settings(_.withIncomingConnectionLevelBufferSize(ConnectionBufferSize))
+
+ // get the request dispatched and both windows replenished to
their configured sizes
+ network.sendDATA(TheStreamId, endStream = false,
ByteString("0000"))
+ entityDataIn.expectUtf8EncodedString("0000")
+ network.pollForWindowUpdates(500.millis)
+
+ // a peer ignoring the stream-level window: the stream is reset
but the connection keeps running
+ network.sendFrame(DataFrame(TheStreamId, endStream = false,
ByteString("0" * OversizedFrameSize)))
+ network.updateWindowForIncomingDataOnConnection(_ -
OversizedFrameSize) // sendFrame bypasses the tracking
+ network.expectRST_STREAM(TheStreamId, ErrorCode.FLOW_CONTROL_ERROR)
+
+ // the dropped data does not stay reserved, so the peer gets its
whole connection window back
+ network.pollForWindowUpdates(500.millis)
+ network.remainingWindowForIncomingDataOnConnection shouldEqual
ConnectionBufferSize
+ })
"fail stream if request entity is not fully pulled when connection
dies".inAssertAllStagesStopped(
new WaitingForRequestData {
network.sendDATA(TheStreamId, endStream = false,
ByteString("0000"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
