From c70ec0ff03a2c760bbbe26b87a0bb7a45fc1c16e Mon Sep 17 00:00:00 2001 From: Vamsi Vaddavalli Date: Mon, 17 Aug 2026 22:42:36 -0500 Subject: [PATCH] Ignore DATA and HEADERS frames received after stream reset Track recently reset stream IDs in a bounded set so late in-flight frames on canceled or reset streams are discarded instead of triggering an erroneous RST_STREAM with PROTOCOL_ERROR. Leftover DATA bytes still count toward connection-level flow control. --- .../okhttp3/internal/http2/Http2Connection.kt | 55 +++++- .../okhttp3/internal/http2/Http2Stream.kt | 3 + .../internal/http2/Http2ConnectionTest.kt | 161 +++++++++++++++++- 3 files changed, 203 insertions(+), 16 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..fcfc16b3a661 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt @@ -144,6 +144,9 @@ class Http2Connection internal constructor( // Guarded by this. private val currentPushRequests = mutableSetOf() + // Streams we've reset so we can ignore late-arriving frames. + private val resetStreamIds = linkedSetOf() + init { if (builder.pingIntervalMillis != 0) { val pingIntervalNanos = TimeUnit.MILLISECONDS.toNanos(builder.pingIntervalMillis.toLong()) @@ -186,6 +189,19 @@ class Http2Connection internal constructor( } } + internal fun recordRstStream(streamId: Int) { + withLock { + if (!resetStreamIds.add(streamId)) return + if (resetStreamIds.size > MAX_TRACKED_RESET_STREAM_IDS) { + val eldest = resetStreamIds.iterator() + eldest.next() + eldest.remove() + } + } + } + + internal fun wasReset(streamId: Int): Boolean = withLock { streamId in resetStreamIds } + internal fun updateConnectionFlowControl(read: Long) { withLock { readBytes.update(total = read) @@ -339,6 +355,7 @@ class Http2Connection internal constructor( streamId: Int, errorCode: ErrorCode, ) { + recordRstStream(streamId) writerQueue.execute("$connectionName[$streamId] writeSynReset") { try { writeSynReset(streamId, errorCode) @@ -353,6 +370,7 @@ class Http2Connection internal constructor( streamId: Int, statusCode: ErrorCode, ) { + recordRstStream(streamId) writer.rstStream(streamId, statusCode) } @@ -647,17 +665,31 @@ class Http2Connection internal constructor( source: BufferedSource, length: Int, ) { + val dataStream: Http2Stream? + val rstSent: Boolean + withLock { + dataStream = streams[streamId] + rstSent = streamId in resetStreamIds + } + + if (rstSent) { + updateConnectionFlowControl(length.toLong()) + source.skip(length.toLong()) + return + } + if (pushedStream(streamId)) { pushDataLater(streamId, source, length, inFinished) return } - val dataStream = getStream(streamId) + if (dataStream == null) { writeSynResetLater(streamId, ErrorCode.PROTOCOL_ERROR) updateConnectionFlowControl(length.toLong()) source.skip(length.toLong()) return } + dataStream.receiveData(source, length) if (inFinished) { dataStream.receiveHeaders(Headers.EMPTY, true) @@ -670,6 +702,8 @@ class Http2Connection internal constructor( associatedStreamId: Int, headerBlock: List
, ) { + if (wasReset(streamId)) return + if (pushedStream(streamId)) { pushHeadersLater(streamId, headerBlock, inFinished) return @@ -872,10 +906,8 @@ class Http2Connection internal constructor( } } else { val stream = getStream(streamId) - if (stream != null) { - stream.withLock { - stream.addBytesToWriteWindow(windowSizeIncrement) - } + stream?.withLock { + stream.addBytesToWriteWindow(windowSizeIncrement) } } } @@ -917,7 +949,7 @@ class Http2Connection internal constructor( requestHeaders: List
, ) { withLock { - if (streamId in currentPushRequests) { + if (streamId in currentPushRequests || streamId in resetStreamIds) { writeSynResetLater(streamId, ErrorCode.PROTOCOL_ERROR) return } @@ -927,7 +959,7 @@ class Http2Connection internal constructor( val cancel = pushObserver.onRequest(streamId, requestHeaders) ignoreIoExceptions { if (cancel) { - writer.rstStream(streamId, ErrorCode.CANCEL) + writeSynReset(streamId, ErrorCode.CANCEL) withLock { currentPushRequests.remove(streamId) } @@ -942,9 +974,10 @@ class Http2Connection internal constructor( inFinished: Boolean, ) { pushQueue.execute("$connectionName[$streamId] onHeaders") { + if (wasReset(streamId)) return@execute val cancel = pushObserver.onHeaders(streamId, requestHeaders, inFinished) ignoreIoExceptions { - if (cancel) writer.rstStream(streamId, ErrorCode.CANCEL) + if (cancel) writeSynReset(streamId, ErrorCode.CANCEL) if (cancel || inFinished) { withLock { currentPushRequests.remove(streamId) @@ -969,9 +1002,11 @@ class Http2Connection internal constructor( source.require(byteCount.toLong()) // Eagerly read the frame before firing client thread. source.read(buffer, byteCount.toLong()) pushQueue.execute("$connectionName[$streamId] onData") { + updateConnectionFlowControl(byteCount.toLong()) + if (wasReset(streamId)) return@execute ignoreIoExceptions { val cancel = pushObserver.onData(streamId, buffer, byteCount, inFinished) - if (cancel) writer.rstStream(streamId, ErrorCode.CANCEL) + if (cancel) writeSynReset(streamId, ErrorCode.CANCEL) if (cancel || inFinished) { withLock { currentPushRequests.remove(streamId) @@ -1042,5 +1077,7 @@ class Http2Connection internal constructor( const val DEGRADED_PING = 2 const val AWAIT_PING = 3 const val DEGRADED_PONG_TIMEOUT_NS = 1_000_000_000 // 1 second. + + const val MAX_TRACKED_RESET_STREAM_IDS = 256 } } diff --git a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt index 2c1b1779023a..7a915274f2ae 100644 --- a/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt +++ b/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Stream.kt @@ -282,6 +282,9 @@ class Http2Stream internal constructor( return false } } + + // Record the reset before removing the stream so the reader thread never sees an unknown stream. + connection.recordRstStream(id) connection.removeStream(id) return true } diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt index a073f1d41b61..cd8e8319c79e 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt +++ b/okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt @@ -191,8 +191,7 @@ 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. + * Leftover DATA in flight when we cancelled must still count toward the connection flow-control window. */ @Test fun discardedDataFramesAreCounted() { // Write the mocking script. @@ -202,8 +201,11 @@ class Http2ConnectionTest { peer.sendFrame().headers(false, 3, headerEntries("a", "apple")) peer.sendFrame().data(false, 3, data(1024), 1024) peer.acceptFrame() // RST_STREAM + peer.sendFrame().data(false, 3, data(1024), 1024) + peer.sendFrame().data(false, 3, data(1024), 1024) peer.sendFrame().data(true, 3, data(1024), 1024) - peer.acceptFrame() // RST_STREAM + peer.sendFrame().ping(false, 2, 0) + peer.acceptFrame() // PING peer.play() val connection = connect(peer) val stream1 = connection.newStream(headerEntries("b", "bark"), false) @@ -215,10 +217,92 @@ class Http2ConnectionTest { assertThat(frame1.type).isEqualTo(Http2.TYPE_HEADERS) val frame2 = peer.takeFrame() assertThat(frame2.type).isEqualTo(Http2.TYPE_RST_STREAM) - val frame3 = peer.takeFrame() - assertThat(frame3.type).isEqualTo(Http2.TYPE_RST_STREAM) + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.payload1).isEqualTo(2) assertThat(connection.readBytes.acknowledged).isEqualTo(0L) - assertThat(connection.readBytes.total).isEqualTo(2048L) + assertThat(connection.readBytes.total).isEqualTo(4096L) + } + + @Test fun dataFrameAfterAsyncCancelIsIgnored() { + // Write the mocking script. + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // SYN_STREAM + peer.sendFrame().headers(false, 3, headerEntries("a", "apple")) + peer.acceptFrame() // RST_STREAM + peer.sendFrame().data(true, 3, data(1024), 1024) + peer.sendFrame().ping(false, 2, 0) + peer.acceptFrame() // PING + peer.play() + + // Play it back. + val connection = connect(peer) + val stream = connection.newStream(headerEntries("b", "bark"), false) + stream.cancel() + + // The late DATA must not trigger another RST_STREAM. + val frame1 = peer.takeFrame() + assertThat(frame1.type).isEqualTo(Http2.TYPE_HEADERS) + val frame2 = peer.takeFrame() + assertThat(frame2.type).isEqualTo(Http2.TYPE_RST_STREAM) + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.payload1).isEqualTo(2) + assertThat(connection.readBytes.total).isEqualTo(1024L) + } + + @Test fun headersFrameAfterCancelIsIgnored() { + // Write the mocking script. + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.acceptFrame() // SYN_STREAM + peer.sendFrame().headers(false, 3, headerEntries("a", "apple")) + peer.acceptFrame() // RST_STREAM + peer.sendFrame().headers(true, 3, headerEntries("trailer", "peach")) + peer.sendFrame().ping(false, 2, 0) + peer.acceptFrame() // PING + peer.play() + + // Play it back. + val connection = connect(peer) + val stream = connection.newStream(headerEntries("b", "bark"), false) + stream.cancel() + + // The late HEADERS must not trigger another RST_STREAM. + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_HEADERS) + assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_RST_STREAM) + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.payload1).isEqualTo(2) + } + + @Test fun trackedResetStreamIdsAreBounded() { + peer.play() + val connection = newConnection(taskFaker.taskRunner) + + val eldestStreamId = 3 + val newestStreamId = eldestStreamId + 2 * Http2Connection.MAX_TRACKED_RESET_STREAM_IDS + for (streamId in eldestStreamId..newestStreamId step 2) { + connection.recordRstStream(streamId) + } + + assertThat(connection.wasReset(eldestStreamId)).isFalse() + assertThat(connection.wasReset(eldestStreamId + 2)).isTrue() + assertThat(connection.wasReset(newestStreamId)).isTrue() + } + + @Test fun resettingSameStreamTwiceDoesNotEvict() { + peer.play() + val connection = newConnection(taskFaker.taskRunner) + + connection.recordRstStream(3) + for (i in 0..Http2Connection.MAX_TRACKED_RESET_STREAM_IDS) { + connection.recordRstStream(5) + } + + assertThat(connection.wasReset(3)).isTrue() + assertThat(connection.wasReset(5)).isTrue() } @Test fun receiveGoAwayHttp2() { @@ -470,16 +554,72 @@ class Http2ConnectionTest { ), ) peer.acceptFrame() // RST_STREAM + peer.sendFrame().data(true, 2, data(1024), 1024) + peer.sendFrame().ping(false, 2, 0) + peer.acceptFrame() // PING peer.play() // Play it back. - connect(peer, PushObserver.CANCEL, Http2Connection.Listener.REFUSE_INCOMING_STREAMS) + val connection = + connect(peer, PushObserver.CANCEL, Http2Connection.Listener.REFUSE_INCOMING_STREAMS) // Verify the peer received what was expected. val rstStream = peer.takeFrame() assertThat(rstStream.type).isEqualTo(Http2.TYPE_RST_STREAM) assertThat(rstStream.streamId).isEqualTo(2) assertThat(rstStream.errorCode).isEqualTo(ErrorCode.CANCEL) + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.payload1).isEqualTo(2) + assertThat(connection.readBytes.total).isEqualTo(1024L) + } + + @Test fun pushPromiseForResetStreamIsRejected() { + // Write the mocking script. + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.sendFrame().pushPromise(3, 2, headerEntries("a", "apple")) + peer.acceptFrame() // RST_STREAM + peer.sendFrame().pushPromise(3, 2, headerEntries("b", "banana")) + peer.acceptFrame() // RST_STREAM + peer.sendFrame().pushPromise(3, 2, headerEntries("c", "cherry")) + peer.acceptFrame() // RST_STREAM + peer.play() + + // Play it back. + connect(peer, PushObserver.CANCEL, Http2Connection.Listener.REFUSE_INCOMING_STREAMS) + + // The push observer cancels the first promise. + val cancel = peer.takeFrame() + assertThat(cancel.type).isEqualTo(Http2.TYPE_RST_STREAM) + assertThat(cancel.streamId).isEqualTo(2) + assertThat(cancel.errorCode).isEqualTo(ErrorCode.CANCEL) + + repeat(2) { + val rstStream = peer.takeFrame() + assertThat(rstStream.type).isEqualTo(Http2.TYPE_RST_STREAM) + assertThat(rstStream.streamId).isEqualTo(2) + assertThat(rstStream.errorCode).isEqualTo(ErrorCode.PROTOCOL_ERROR) + } + } + + @Test fun pushedDataIsCountedAgainstConnectionFlowControl() { + // Write the mocking script. + peer.sendFrame().settings(Settings()) + peer.acceptFrame() // ACK + peer.sendFrame().pushPromise(3, 2, headerEntries("a", "apple")) + peer.sendFrame().data(true, 2, data(1024), 1024) + peer.sendFrame().ping(false, 2, 0) + peer.acceptFrame() // PING + peer.play() + + // Play it back. + val connection = connect(peer) + + val ping = peer.takeFrame() + assertThat(ping.type).isEqualTo(Http2.TYPE_PING) + assertThat(ping.payload1).isEqualTo(2) + assertThat(connection.readBytes.total).isEqualTo(1024L) } /** @@ -1986,6 +2126,13 @@ class Http2ConnectionTest { return connection } + private fun newConnection(taskRunner: TaskRunner): Http2Connection = + Http2Connection + .Builder(true, taskRunner) + .socket(peer.openSocket().asBufferedSocket(), "peer") + .pushObserver(IGNORE) + .build() + private class RecordingPushObserver : PushObserver, Lockable {