Ignore DATA and HEADERS frames received after stream reset

okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/http2/Http2Connection.kt

@@ -144,6 +144,9 @@ class Http2Connection internal constructor(
   // Guarded by this.
   private val currentPushRequests = mutableSetOf<Int>()
 
+  // Streams we've reset so we can ignore late-arriving frames.
+  private val resetStreamIds = linkedSetOf<Int>()
+
   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<Header>,
     ) {
+      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<Header>,
   ) {
     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
   }
 }

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
   }

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 {