From 983d9f6ef977d69c68d7cd5ded2dc865c327250a Mon Sep 17 00:00:00 2001 From: Amir Golmoradi Date: Mon, 12 Oct 2026 22:57:53 +0330 Subject: [PATCH 1/2] Count DATA frame padding toward HTTP/2 flow control Fixes #9772 --- .../okhttp3/internal/http2/Http2Connection.kt | 8 + .../okhttp3/internal/http2/Http2Reader.kt | 9 ++ .../okhttp3/internal/http2/Http2Stream.kt | 29 +++- .../okhttp3/internal/http2/BaseTestHandler.kt | 7 + .../internal/http2/Http2ConnectionTest.kt | 151 ++++++++++++++++++ .../okhttp3/internal/http2/Http2Test.kt | 33 +++- .../okhttp3/internal/http2/MockHttp2Peer.kt | 26 +++ 7 files changed, 254 insertions(+), 9 deletions(-) diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt index b79401a79f64..fa0732811ded 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt @@ -664,6 +664,14 @@ class Http2Connection internal constructor( } } + override fun dataPadding( + streamId: Int, + length: Int, + ) { + updateConnectionFlowControl(length.toLong()) + getStream(streamId)?.source?.receivePadding(length.toLong()) + } + override fun headers( inFinished: Boolean, streamId: Int, diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Reader.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Reader.kt index bc3732ec4989..e86c66126eec 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Reader.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Reader.kt @@ -200,6 +200,9 @@ class Http2Reader( handler.data(inFinished, streamId, source, dataLength) source.skip(padding.toLong()) + if (length > dataLength) { + handler.dataPadding(streamId, length - dataLength) + } } @Throws(IOException::class) @@ -458,6 +461,12 @@ class Http2Reader( length: Int, ) + /** Reports discarded DATA padding, including the Pad Length field. */ + fun dataPadding( + streamId: Int, + length: Int, + ) + /** * Create or update incoming headers, creating the corresponding streams if necessary. Frames * that trigger this are HEADERS and PUSH_PROMISE. diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt index 2c1b1779023a..cfb42ea36fa4 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt @@ -405,14 +405,10 @@ class Http2Stream internal constructor( readBytesDelivered = readBuffer.read(sink, minOf(byteCount, readBuffer.size)) readBytes.update(total = readBytesDelivered) - val unacknowledgedBytesRead = readBytes.unacknowledged - if (errorExceptionToDeliver == null && - unacknowledgedBytesRead >= connection.okHttpSettings.initialWindowSize / 2 - ) { + if (errorExceptionToDeliver == null) { // Flow control: notify the peer that we're ready for more data! Only send a // WINDOW_UPDATE if the stream isn't in error. - connection.writeWindowUpdateLater(id, unacknowledgedBytesRead) - readBytes.update(acknowledged = unacknowledgedBytesRead) + acknowledgeBytesRead() } } else if (!finished && errorExceptionToDeliver == null) { // Nothing to do. Wait until that changes then try again. @@ -449,6 +445,27 @@ class Http2Stream internal constructor( } } + private fun acknowledgeBytesRead() { + val unacknowledgedBytesRead = readBytes.unacknowledged + if (unacknowledgedBytesRead >= connection.okHttpSettings.initialWindowSize / 2) { + connection.writeWindowUpdateLater(id, unacknowledgedBytesRead) + readBytes.update(acknowledged = unacknowledgedBytesRead) + } + } + + /** Padding is consumed by the reader, so it can be acknowledged without an application read. */ + internal fun receivePadding(byteCount: Long) { + assertLockNotHeld() + + withLock { + readBytes.update(total = byteCount) + if (!closed && errorCode == null) { + acknowledgeBytesRead() + } + } + connection.flowControlListener.receivingStreamWindowChanged(id, readBytes, readBuffer.size) + } + private fun updateConnectionFlowControl(read: Long) { assertLockNotHeld() diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/BaseTestHandler.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/BaseTestHandler.kt index 4c70d410767d..78d5bc7be52f 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/BaseTestHandler.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/BaseTestHandler.kt @@ -29,6 +29,13 @@ internal open class BaseTestHandler : Http2Reader.Handler { fail("") } + override fun dataPadding( + streamId: Int, + length: Int, + ) { + fail("") + } + override fun headers( inFinished: Boolean, streamId: Int, diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt index a073f1d41b61..339b38022130 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt @@ -54,6 +54,8 @@ import org.junit.jupiter.api.Assertions.assertArrayEquals import org.junit.jupiter.api.Tag import org.junit.jupiter.api.Test import org.junit.jupiter.api.Timeout +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource @Timeout(5) @Tag("Slow") @@ -318,6 +320,155 @@ class Http2ConnectionTest { } } + @ParameterizedTest + @ValueSource(ints = [0, 39, 255]) + fun paddedDataSendsConnectionWindowUpdate(padding: Int) { + val payloadLength = 10 + 1 + padding + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // HEADERS + peer.sendFrame().headers(false, 3, headerEntries("a", "android")) + peer.sendPaddedData(true, 3, data(10), 10, padding) + peer.acceptFrame() // Connection WINDOW_UPDATE, before the application reads the body. + peer.sendFrame().ping(false, 2, 3) + peer.acceptFrame() // PING ACK + peer.play() + + val connection = connect(peer) + connection.okHttpSettings[Settings.INITIAL_WINDOW_SIZE] = payloadLength * 2 + val stream = connection.newStream(headerEntries("b", "banana"), false) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + val windowUpdate = peer.takeFrame() + assertThat(windowUpdate.type).isEqualTo(Http2.TYPE_WINDOW_UPDATE) + assertThat(windowUpdate.streamId).isEqualTo(0) + assertThat(windowUpdate.windowSizeIncrement).isEqualTo(payloadLength.toLong()) + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.ack).isTrue() + assertThat(connection.readBytes.total).isEqualTo(payloadLength.toLong()) + assertThat(connection.readBytes.acknowledged).isEqualTo(payloadLength.toLong()) + val buffer = Buffer() + buffer.writeAll(stream.source) + assertThat(buffer).isEqualTo(data(10)) + } + + @Test fun paddedDataSendsStreamWindowUpdate() { + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // HEADERS + peer.sendFrame().headers(false, 3, headerEntries("a", "android")) + repeat(3) { + peer.sendPaddedData(false, 3, data(10), 10, 39) + peer.acceptFrame() // Connection WINDOW_UPDATE + peer.acceptFrame() // Stream WINDOW_UPDATE + } + peer.sendFrame().data(true, 3, data(0), 0) + peer.play() + + val connection = connect(peer) + connection.okHttpSettings[Settings.INITIAL_WINDOW_SIZE] = 100 + val stream = connection.newStream(headerEntries("b", "banana"), false) + val buffer = Buffer() + buffer.writeAll(stream.source) + assertThat(buffer).isEqualTo(data(30)) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + repeat(3) { + val updates = listOf(peer.takeFrame(), peer.takeFrame()) + assertThat(updates.map { it.streamId }.toSet()).isEqualTo(setOf(0, 3)) + for (update in updates) { + assertThat(update.type).isEqualTo(Http2.TYPE_WINDOW_UPDATE) + assertThat(update.windowSizeIncrement).isEqualTo(50L) + } + } + assertThat(connection.readBytes.total).isEqualTo(150L) + assertThat(connection.readBytes.acknowledged).isEqualTo(150L) + assertThat(stream.readBytes.total).isEqualTo(150L) + assertThat(stream.readBytes.acknowledged).isEqualTo(150L) + } + + @Test fun paddingOnlyDataSendsWindowUpdateWithoutReadingBody() { + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // HEADERS + peer.sendFrame().headers(false, 3, headerEntries("a", "android")) + peer.sendPaddedData(false, 3, data(0), 0, 49) + peer.acceptFrame() // Connection WINDOW_UPDATE + peer.acceptFrame() // Stream WINDOW_UPDATE + peer.sendFrame().data(true, 3, data(0), 0) + peer.play() + + val connection = connect(peer) + connection.okHttpSettings[Settings.INITIAL_WINDOW_SIZE] = 100 + val stream = connection.newStream(headerEntries("b", "banana"), false) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + val updates = listOf(peer.takeFrame(), peer.takeFrame()) + assertThat(updates.map { it.streamId }.toSet()).isEqualTo(setOf(0, 3)) + for (update in updates) { + assertThat(update.type).isEqualTo(Http2.TYPE_WINDOW_UPDATE) + assertThat(update.windowSizeIncrement).isEqualTo(50L) + } + assertThat(stream.source.read(Buffer(), 1)).isEqualTo(-1L) + assertThat(connection.readBytes.total).isEqualTo(50L) + assertThat(stream.readBytes.total).isEqualTo(50L) + } + + @ParameterizedTest + @ValueSource(booleans = [false, true]) + fun paddedDataOnUnknownStreamSendsConnectionWindowUpdate(removed: Boolean) { + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + if (removed) { + peer.acceptFrame() // HEADERS + peer.sendFrame().rstStream(3, ErrorCode.CANCEL) + } + peer.acceptFrame() // PING + peer.sendPaddedData(true, 3, data(10), 10, 39) + peer.acceptFrame() // RST_STREAM + peer.acceptFrame() // Connection WINDOW_UPDATE + peer.sendFrame().ping(true, Http2Connection.AWAIT_PING, 0) + peer.play() + + val connection = connect(peer) + connection.okHttpSettings[Settings.INITIAL_WINDOW_SIZE] = 100 + if (removed) { + connection.newStream(headerEntries("b", "banana"), false) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + } + connection.writePingAndAwaitPong() + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_PING) + val reset = peer.takeFrame() + assertThat(reset.type).isEqualTo(Http2.TYPE_RST_STREAM) + assertThat(reset.streamId).isEqualTo(3) + val windowUpdate = peer.takeFrame() + assertThat(windowUpdate.type).isEqualTo(Http2.TYPE_WINDOW_UPDATE) + assertThat(windowUpdate.streamId).isEqualTo(0) + assertThat(windowUpdate.windowSizeIncrement).isEqualTo(50L) + assertThat(connection.readBytes.total).isEqualTo(50L) + } + + @Test fun paddingOnPushedStreamSendsConnectionWindowUpdate() { + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // HEADERS + peer.sendFrame().headers(false, 3, headerEntries("a", "android")) + peer.sendFrame().pushPromise(3, 2, headerEntries("a", "android")) + peer.sendPaddedData(true, 2, data(0), 0, 49) + peer.acceptFrame() // Connection WINDOW_UPDATE + peer.sendFrame().data(true, 3, data(0), 0) + peer.play() + + val connection = connect(peer) + connection.okHttpSettings[Settings.INITIAL_WINDOW_SIZE] = 100 + val stream = connection.newStream(headerEntries("b", "banana"), false) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + val windowUpdate = peer.takeFrame() + assertThat(windowUpdate.type).isEqualTo(Http2.TYPE_WINDOW_UPDATE) + assertThat(windowUpdate.streamId).isEqualTo(0) + assertThat(windowUpdate.windowSizeIncrement).isEqualTo(50L) + assertThat(stream.source.read(Buffer(), 1)).isEqualTo(-1L) + assertThat(connection.readBytes.total).isEqualTo(50L) + } + @Test fun serverSendsEmptyDataClientDoesntSendWindowUpdateHttp2() { // Write the mocking script. peer.sendFrame().settings(Settings()) diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2Test.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2Test.kt index 5e8c5789691c..133c2fa4113e 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2Test.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2Test.kt @@ -515,7 +515,9 @@ class Http2Test { frame.writeByte(paddingLength) frame.write(expectedData) frame.write(padding) - reader.nextFrame(requireSettings = false, assertData()) + val paddingBytes = AtomicInteger() + reader.nextFrame(requireSettings = false, assertData(paddingBytes)) + assertThat(paddingBytes.get()).isEqualTo(paddingLength + 1) // Padding was skipped. assertThat(frame.exhausted()).isTrue() } @@ -530,7 +532,24 @@ class Http2Test { frame.writeInt(expectedStreamId and 0x7fffffff) frame.writeByte(0) frame.write(expectedData) - reader.nextFrame(requireSettings = false, assertData()) + val paddingBytes = AtomicInteger() + reader.nextFrame(requireSettings = false, assertData(paddingBytes)) + assertThat(paddingBytes.get()).isEqualTo(1) + } + + @Test fun truncatedDataPaddingIsNotReported() { + writeMedium(frame, 1123 + 1 + 39) + frame.writeByte(Http2.TYPE_DATA) + frame.writeByte(FLAG_PADDED) + frame.writeInt(expectedStreamId) + frame.writeByte(39) + frame.write(ByteArray(1123) { 2 }) + frame.write(ByteArray(38)) + val paddingBytes = AtomicInteger() + assertFailsWith { + reader.nextFrame(requireSettings = false, assertData(paddingBytes)) + } + assertThat(paddingBytes.get()).isEqualTo(0) } @Test fun readPaddedHeadersFrame() { @@ -795,7 +814,7 @@ class Http2Test { } } - private fun assertData(): Http2Reader.Handler = + private fun assertData(paddingBytes: AtomicInteger = AtomicInteger()): Http2Reader.Handler = object : BaseTestHandler() { override fun data( inFinished: Boolean, @@ -811,6 +830,14 @@ class Http2Test { assertThat(b).isEqualTo(2.toByte()) } } + + override fun dataPadding( + streamId: Int, + length: Int, + ) { + assertThat(streamId).isEqualTo(expectedStreamId) + paddingBytes.addAndGet(length) + } } private fun gzip(data: ByteArray): Buffer { diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/MockHttp2Peer.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/MockHttp2Peer.kt index ae992eed972a..8931631ed545 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/MockHttp2Peer.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/MockHttp2Peer.kt @@ -66,6 +66,21 @@ class MockHttp2Peer : Closeable { return writer } + fun sendPaddedData( + inFinished: Boolean, + streamId: Int, + data: Buffer, + byteCount: Int, + padding: Int, + ) { + require(padding in 0..255) + val flags = Http2.FLAG_PADDED or if (inFinished) Http2.FLAG_END_STREAM else 0 + sendFrame().frameHeader(streamId, byteCount + padding + 1, Http2.TYPE_DATA, flags) + bytesOut.writeByte(padding) + bytesOut.write(data, byteCount.toLong()) + bytesOut.write(ByteArray(padding)) + } + /** * Shortens the last frame from its original length to `length`. This will cause the peer to * close the socket as soon as this frame has been written; otherwise the peer stays open until @@ -195,6 +210,8 @@ class MockHttp2Peer : Closeable { @JvmField var data: ByteArray? = null + @JvmField var padding = 0 + @JvmField var settings: Settings? = null @JvmField var ack = false @@ -246,6 +263,15 @@ class MockHttp2Peer : Closeable { this.data = source.readByteString(length.toLong()).toByteArray() } + override fun dataPadding( + streamId: Int, + length: Int, + ) { + check(type == Http2.TYPE_DATA) + check(this.streamId == streamId) + this.padding = length + } + override fun rstStream( streamId: Int, errorCode: ErrorCode, From 939dc09aec3ebaf873988103cadc4e2b88c4dae5 Mon Sep 17 00:00:00 2001 From: Amir Golmoradi Date: Thu, 22 Oct 2026 16:57:55 +0330 Subject: [PATCH 2/2] Fix race in discardedDataFramesAreCounted test --- .../okhttp3/internal/http2/Http2ConnectionTest.kt | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt index 339b38022130..cae668c3d806 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt @@ -196,7 +196,8 @@ class Http2ConnectionTest { * Confirm that we account for discarded data frames. It's possible that data frames are in-flight * just prior to us canceling a stream. */ - @Test fun discardedDataFramesAreCounted() { + @Test + fun discardedDataFramesAreCounted() { // Write the mocking script. peer.sendFrame().settings(Settings()) peer.acceptFrame() // ACK @@ -206,6 +207,8 @@ class Http2ConnectionTest { peer.acceptFrame() // RST_STREAM peer.sendFrame().data(true, 3, data(1024), 1024) peer.acceptFrame() // RST_STREAM + peer.sendFrame().ping(false, 2, 3) + peer.acceptFrame() // PING ACK peer.play() val connection = connect(peer) val stream1 = connection.newStream(headerEntries("b", "bark"), false) @@ -219,6 +222,14 @@ class Http2ConnectionTest { assertThat(frame2.type).isEqualTo(Http2.TYPE_RST_STREAM) val frame3 = peer.takeFrame() assertThat(frame3.type).isEqualTo(Http2.TYPE_RST_STREAM) + + // Wait until the reader has finished processing the discarded DATA frame. + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.ack).isTrue() + assertThat(ping.payload1).isEqualTo(2) + assertThat(ping.payload2).isEqualTo(3) + assertThat(connection.readBytes.acknowledged).isEqualTo(0L) assertThat(connection.readBytes.total).isEqualTo(2048L) }