From cd56042c98c231481405f12d035a3b24d0a41a9b Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Mon, 31 Aug 2026 16:49:02 +0100 Subject: [PATCH] fix: release connection-level flow control accounting for discarded buffered data 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 with the fix stashed (the peer runs out of connection window). - 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 --- .../engine/http2/Http2StreamHandling.scala | 18 ++++++++++++--- .../impl/engine/http2/Http2ServerSpec.scala | 22 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 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 5de3aa173..364e523fc 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 @@ -391,7 +391,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)) { @@ -640,10 +642,20 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper streamStates.remove(streamId) headRequestStreamIds -= 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 = @@ -676,7 +688,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 } 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 78bd9d78b..c90661b85 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 @@ -346,6 +346,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 =