This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.4.x
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/1.4.x by this push:
new fc2c965dc fix: release HTTP/2 connection-level buffered data
accounting on discard (#1259, #1281) (1.4.x) (#1328)
fc2c965dc is described below
commit fc2c965dcc3fc35afc7d198fb5c8098583024606
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 9 20:00:53 2026 +0100
fix: release HTTP/2 connection-level buffered data accounting on discard
(#1259, #1281) (1.4.x) (#1328)
* fix: release connection-level flow control accounting for discarded
buffered data (#1259)
Motivation:
`totalBufferedData` feeds the connection-level flow controller, which
emits a WINDOW_UPDATE only while `outstanding + buffered` stays below
half of `incoming-connection-level-buffer-size`. It is incremented for
every DATA frame received but was not decremented on the three paths
that discard a non-empty buffer:
- `IncomingStreamBuffer.onRstStreamFrame` cleared the buffer after the
peer reset the stream,
- `IncomingStreamBuffer.onDownstreamFinish` cleared it when the
application cancelled the entity stream,
- `CollectingIncomingData.onRstStreamFrame` was a no-op even though the
bytes collected so far had been counted.
A peer that sends data the handler does not read and then resets the
stream therefore ratchets `totalBufferedData` up permanently. Once the
leaked total reaches half the configured buffer size the server stops
replenishing the connection window, it drains to zero and every stream
on that connection stalls.
Modification:
Add `IncomingStreamBuffer.discardBuffer()`, which subtracts what is
still buffered from `totalBufferedData` before clearing it, and use it
on both discard paths. Subtract the collected bytes in
`CollectingIncomingData.onRstStreamFrame` for the same reason.
Result:
Resetting or cancelling a stream releases the connection-level window
its buffered data reserved, so the connection keeps being replenished
and no longer stalls.
Tests:
- sbt "http2-tests/testOnly
org.apache.pekko.http.impl.engine.http2.Http2ServerSpec
org.apache.pekko.http.impl.engine.http2.Http2ClientSpec
org.apache.pekko.http.impl.engine.http2.Http2ClientServerSpec" - pass (174
tests); a new test buffers request data the handler never reads and resets the
stream six times over, with the connection-level buffer size lowered to the
initial window so the effect is reached quickly, then asserts the connection
still accepts a request. Verified it fails [...]
- sbt http-core/mimaReportBinaryIssues - pass (internal impl.engine.http2
change, no public API).
References:
None - releases buffered-data accounting when a stream is reset or cancelled
(cherry picked from commit 2dc91fef995e89564186fcdf927dccde2193f462)
* 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
(cherry picked from commit 83eb9486df341dfaf22ee8d9a37148c1785cb330)
---
.../impl/engine/http2/Http2StreamHandling.scala | 31 ++++++++++++--
.../http/impl/engine/http2/Http2ServerSpec.scala | 47 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 4 deletions(-)
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 bbd1d1540..2394cf425 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
@@ -371,7 +371,9 @@ private[http2] trait Http2StreamHandling extends
GraphStageLogic with LogHelper
override protected def onTrailer(parsedHeadersFrame: ParsedHeadersFrame):
StreamState = this // trailing headers not supported for requests right now
override protected def incrementWindow(delta: Int): StreamState =
copy(extraInitialWindow = extraInitialWindow + delta)
- override protected def onRstStreamFrame(rstStreamFrame: RstStreamFrame):
Unit = {} // nothing to do here
+ // the data collected so far is dropped together with this state, so stop
reserving connection-level window for it
+ override protected def onRstStreamFrame(rstStreamFrame: RstStreamFrame):
Unit =
+ totalBufferedData -= collectedData.length
}
case class OpenReceivingDataFirst(buffer: IncomingStreamBuffer,
extraInitialWindow: Int = 0)
extends
ReceivingDataWithBuffer(HalfClosedRemoteWaitingForOutgoingStream(extraInitialWindow))
{
@@ -604,10 +606,20 @@ private[http2] trait Http2StreamHandling extends
GraphStageLogic with LogHelper
// FIXME: go through state machine and don't manipulate vars directly
here
streamStates.remove(streamId)
wasClosed = true
- buffer = ByteString.empty
+ discardBuffer()
trailingHeaders = None
}
+ /**
+ * Drops what is still buffered and releases the connection-level window
it reserved. Without the release the
+ * connection-level flow controller keeps counting these bytes as buffered
forever, so it stops replenishing the
+ * connection window and every stream on the connection eventually stalls.
+ */
+ private def discardBuffer(): Unit = {
+ totalBufferedData -= buffer.length
+ buffer = ByteString.empty
+ }
+
def isDone: Boolean = outlet.isClosed
def onDataFrame(data: DataFrame): Unit =
@@ -619,6 +631,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
@@ -640,7 +654,7 @@ private[http2] trait Http2StreamHandling extends
GraphStageLogic with LogHelper
}
def onRstStreamFrame(rst: RstStreamFrame): Unit = {
outlet.fail(new PeerClosedStreamException(rst.streamId, rst.errorCode))
- buffer = ByteString.empty
+ discardBuffer()
trailingHeaders = None
wasClosed = true
}
@@ -692,8 +706,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 027c80731..deb6223fb 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
@@ -370,6 +370,28 @@ class Http2ServerSpec extends Http2SpecWithMaterializer("""
trailingResponseHeaders should contain(("x-good", "fine"))
trailingResponseHeaders.map(_._1) should not contain "x-evil"
})
+ "release connection-level flow control accounting when a stream with
buffered data is reset"
+ .inAssertAllStagesStopped(new TestSetup with RequestResponseProbes {
+ override def settings: ServerSettings =
+
super.settings.mapHttp2Settings(_.withIncomingConnectionLevelBufferSize(Http2Protocol.InitialWindowSize))
+
+ // Each round buffers request data that the handler never reads and
then resets the stream. Those bytes must
+ // be released from the connection-level accounting: otherwise the
flow controller keeps counting them as
+ // buffered forever, stops replenishing the connection window, and
the peer runs out of window entirely.
+ (0 until 6).foreach { i =>
+ val streamId = 1 + i * 2
+ network.sendHEADERS(streamId, endStream = false,
+ network.headersForRequest(HttpRequest(HttpMethods.POST, "/")))
+ user.expectRequest() // the entity is deliberately never read, so
the data stays buffered
+ network.sendDATA(streamId, endStream = false, ByteString(new
Array[Byte](20000)))
+ network.sendRST_STREAM(streamId, ErrorCode.CANCEL)
+ network.pollForWindowUpdates(100.millis)
+ }
+
+ // the connection is still usable because the window was replenished
along the way
+ network.sendHEADERS(13, endStream = true,
network.headersForRequest(Get("/")))
+ user.expectRequest()
+ })
"consider stream as closed after sending out strict response >
WINDOW_SIZE".inAssertAllStagesStopped(
new TestSetup with RequestResponseProbes {
override def settings: ServerSettings =
@@ -636,6 +658,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]
