Fix for stalled streams

okhttp-coroutines/src/jvmTest/kotlin/okhttp3/TwoRequestTest.kt

@@ -0,0 +1,58 @@
+@file:OptIn(ExperimentalCoroutinesApi::class)
+
+package okhttp3
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runTest
+import okhttp3.HttpUrl.Companion.toHttpUrl
+import org.junit.jupiter.api.Test
+
+class TwoRequestTest {
+  val file = "https://storage.googleapis.com/downloads.webmproject.org/av1/exoplayer/bbb-av1-480p.mp4".toHttpUrl()
+
+  val client = OkHttpClient.Builder()
+    .protocols(listOf(Protocol.HTTP_2, Protocol.HTTP_1_1))
+    .build()
+
+  val extraRequests = false
+  val stalledRequest = true
+
+  @Test
+  fun testTwoQueries() = runTest {
+    if (extraRequests) {
+      val callFull = client.newCall(
+        Request(file, Headers.headersOf("Icy-MetaData", "1", "Accept-Encoding", "identity"))
+      )
+      callFull.executeAsync()
+
+      callFull.cancel()
+
+      val callEnd = client.newCall(
+        Request(file, Headers.headersOf("Range", "bytes=37070547-", "Icy-MetaData", "1", "Accept-Encoding", "identity"))
+      )
+      callEnd.executeAsync()
+      callEnd.cancel()
+    }
+
+    if (stalledRequest) {
+      val callStart = client.newCall(
+        Request(file, Headers.headersOf("Range", "bytes=44-", "Icy-MetaData", "1", "Accept-Encoding", "identity"))
+      )
+      val responseStart = callStart.executeAsync()
+      val bodyStart = responseStart.body
+      bodyStart.byteStream().readNBytes(100_000)
+    }
+
+    val callDownload = client.newCall(Request(file))
+    val responseDownload = callDownload.executeAsync()
+    val bodyDownload = responseDownload.body
+
+    while (true) {
+      val bytes = bodyDownload.byteStream().readNBytes(100_000)
+
+      if (bytes.isEmpty()) {
+        break
+      }
+    }
+  }
+}

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

@@ -188,7 +188,7 @@ class Http2Connection internal constructor(builder: Builder) : Closeable {
   @Synchronized internal fun updateConnectionFlowControl(read: Long) {
     readBytesTotal += read
     val readBytesToAcknowledge = readBytesTotal - readBytesAcknowledged
-    if (readBytesToAcknowledge >= okHttpSettings.initialWindowSize / 2) {
+    if (readBytesToAcknowledge >= Http2Stream.windowThreshold(okHttpSettings)) {
       writeWindowUpdateLater(0, readBytesToAcknowledge)
       readBytesAcknowledged += readBytesToAcknowledge
     }

okhttp/src/jvmMain/kotlin/okhttp3/internal/http2/Http2Stream.kt

@@ -390,7 +390,8 @@ class Http2Stream internal constructor(
 
               val unacknowledgedBytesRead = readBytesTotal - readBytesAcknowledged
               if (errorExceptionToDeliver == null &&
-                  unacknowledgedBytesRead >= connection.okHttpSettings.initialWindowSize / 2) {
+                  unacknowledgedBytesRead >= windowThreshold(connection.okHttpSettings)
+              ) {
                 // 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)
@@ -415,8 +416,6 @@ class Http2Stream internal constructor(
         }
 
         if (readBytesDelivered != -1L) {
-          // Update connection.unacknowledgedBytesRead outside the synchronized block.
-          updateConnectionFlowControl(readBytesDelivered)
           return readBytesDelivered
         }
 
@@ -446,41 +445,39 @@ class Http2Stream internal constructor(
     internal fun receive(source: BufferedSource, byteCount: Long) {
       this@Http2Stream.assertThreadDoesntHoldLock()
 
-      var byteCount = byteCount
+      var remainingByteCount = byteCount
 
-      while (byteCount > 0L) {
+      while (remainingByteCount > 0L) {
         val finished: Boolean
         val flowControlError: Boolean
         synchronized(this@Http2Stream) {
           finished = this.finished
-          flowControlError = byteCount + readBuffer.size > maxByteCount
+          flowControlError = remainingByteCount + readBuffer.size > maxByteCount
         }
 
         // If the peer sends more data than we can handle, discard it and close the connection.
         if (flowControlError) {
-          source.skip(byteCount)
+          source.skip(remainingByteCount)
           closeLater(ErrorCode.FLOW_CONTROL_ERROR)
           return
         }
 
         // Discard data received after the stream is finished. It's probably a benign race.
         if (finished) {
-          source.skip(byteCount)
+          source.skip(remainingByteCount)
           return
         }
 
         // Fill the receive buffer without holding any locks.
-        val read = source.read(receiveBuffer, byteCount)
+        val read = source.read(receiveBuffer, remainingByteCount)
         if (read == -1L) throw EOFException()
-        byteCount -= read
+        remainingByteCount -= read
 
         // Move the received data to the read buffer to the reader can read it. If this source has
         // been closed since this read began we must discard the incoming data and tell the
         // connection we've done so.
-        var bytesDiscarded = 0L
         synchronized(this@Http2Stream) {
           if (closed) {
-            bytesDiscarded = receiveBuffer.size
             receiveBuffer.clear()
           } else {
             val wasEmpty = readBuffer.size == 0L
@@ -490,10 +487,13 @@ class Http2Stream internal constructor(
             }
           }
         }
-        if (bytesDiscarded > 0L) {
-          updateConnectionFlowControl(bytesDiscarded)
-        }
       }
+
+      // Update the connection flow control, as this is a shared resource.
+      // Even if our stream doesn't need more data, others might.
+      // But delay updating the stream flow control until that stream has been
+      // consumed
+      updateConnectionFlowControl(byteCount)
     }
 
     override fun timeout(): Timeout = readTimeout
@@ -656,6 +656,8 @@ class Http2Stream internal constructor(
   }
 
   companion object {
+    internal fun windowThreshold(okHttpSettings: Settings) = okHttpSettings.initialWindowSize / 2
+
     internal const val EMIT_BUFFER_SIZE = 16384L
   }