Count DATA frame padding toward HTTP/2 flow control

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,

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.

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()
 

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,

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")
@@ -194,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
@@ -204,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)
@@ -217,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)
   }
@@ -318,6 +331,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())

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<IOException> {
+      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 {