Fix HTTP/2 write timeout ineffective against TCP-level network blackhole

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

@@ -212,7 +212,7 @@ class Http2Connection internal constructor(
     out: Boolean,
   ): Http2Stream {
     check(!client) { "Client cannot push requests." }
-    return newStream(associatedStreamId, requestHeaders, out)
+    return newStream(associatedStreamId, requestHeaders, out, writeTimeoutMillis = 0L)
   }
 
   /**
@@ -225,13 +225,28 @@ class Http2Connection internal constructor(
   fun newStream(
     requestHeaders: List<Header>,
     out: Boolean,
-  ): Http2Stream = newStream(0, requestHeaders, out)
+  ): Http2Stream = newStream(0, requestHeaders, out, writeTimeoutMillis = 0L)
+
+  /**
+   * Like [newStream], but bounds the blocking write of the initial HEADERS frame by
+   * [writeTimeoutMillis]. Without this, that write -- which happens before the returned
+   * [Http2Stream]'s own write timeout is configured by the caller -- can block on the underlying
+   * socket for as long as the OS takes to give up on an unresponsive peer (see
+   * square/okhttp#9237), ignoring [okhttp3.OkHttpClient.writeTimeoutMillis] entirely.
+   */
+  @Throws(IOException::class)
+  internal fun newStream(
+    requestHeaders: List<Header>,
+    out: Boolean,
+    writeTimeoutMillis: Long,
+  ): Http2Stream = newStream(0, requestHeaders, out, writeTimeoutMillis)
 
   @Throws(IOException::class)
   private fun newStream(
     associatedStreamId: Int,
     requestHeaders: List<Header>,
     out: Boolean,
+    writeTimeoutMillis: Long,
   ): Http2Stream {
     val outFinished = !out
     val inFinished = false
@@ -250,26 +265,33 @@ class Http2Connection internal constructor(
         streamId = nextStreamId
         nextStreamId += 2
         stream = Http2Stream(streamId, this, outFinished, inFinished, null)
+        if (writeTimeoutMillis > 0L) {
+          stream.writeTimeout.timeout(writeTimeoutMillis, TimeUnit.MILLISECONDS)
+        }
         flushHeaders = !out ||
           writeBytesTotal >= writeBytesMaximum ||
           stream.writeBytesTotal >= stream.writeBytesMaximum
         if (stream.isOpen) {
           streams[streamId] = stream
         }
       }
-      if (associatedStreamId == 0) {
-        writer.headers(outFinished, streamId, requestHeaders)
-      } else {
-        require(!client) { "client streams shouldn't have associated stream IDs" }
-        // HTTP/2 has a PUSH_PROMISE frame.
-        writer.pushPromise(associatedStreamId, streamId, requestHeaders)
+      stream.guardConnectionWrite {
+        if (associatedStreamId == 0) {
+          writer.headers(outFinished, streamId, requestHeaders)
+        } else {
+          require(!client) { "client streams shouldn't have associated stream IDs" }
+          // HTTP/2 has a PUSH_PROMISE frame.
+          writer.pushPromise(associatedStreamId, streamId, requestHeaders)
+        }
+        // The write above only fills Http2Writer's internal buffer. flush() is what actually
+        // performs the (potentially blocking) socket write, so it must stay inside this timeout
+        // guard too -- see the class doc on this overload.
+        if (flushHeaders) {
+          writer.flush()
+        }
       }
     }
 
-    if (flushHeaders) {
-      writer.flush()
-    }
-
     return stream
   }
 

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

@@ -81,7 +81,12 @@ class Http2ExchangeCodec(
 
     val hasRequestBody = request.body != null
     val requestHeaders = http2HeadersList(request)
-    stream = http2Connection.newStream(requestHeaders, hasRequestBody)
+    stream =
+      http2Connection.newStream(
+        requestHeaders,
+        hasRequestBody,
+        writeTimeoutMillis = chain.writeTimeoutMillis.toLong(),
+      )
     // We may have been asked to cancel while creating the new stream and sending the request
     // headers, but there was still no stream to close.
     if (canceled) {

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

@@ -78,6 +78,16 @@ class Http2Stream internal constructor(
   internal val readTimeout = StreamTimeout()
   internal val writeTimeout = StreamTimeout()
 
+  /**
+   * True while a thread is parked inside a blocking write to [connection]'s underlying socket
+   * (writing a HEADERS or DATA frame for this stream). Unlike a flow-control wait -- which parks
+   * in [Object.wait] and can always be woken by [closeLater]'s `notifyAll()` -- a thread blocked
+   * inside the actual socket write can only be freed by interrupting the socket itself. See
+   * [StreamTimeout.timedOut].
+   */
+  @Volatile
+  internal var blockedInConnectionWrite = false
+
   /**
    * The reason why this stream was closed, or null if it closed normally or has not yet been
    * closed.
@@ -615,12 +625,7 @@ class Http2Stream internal constructor(
         outFinished = outFinishedOnLastFrame && toWrite == sendBuffer.size
       }
 
-      writeTimeout.enter()
-      try {
-        connection.writeData(id, outFinished, sendBuffer, toWrite)
-      } finally {
-        writeTimeout.exitAndThrowIfTimedOut()
-      }
+      guardConnectionWrite { connection.writeData(id, outFinished, sendBuffer, toWrite) }
     }
 
     @Throws(IOException::class)
@@ -633,7 +638,7 @@ class Http2Stream internal constructor(
       // TODO(jwilson): flush the connection?!
       while (sendBuffer.size > 0L) {
         emitFrame(false)
-        connection.flush()
+        guardConnectionWrite { connection.flush() }
       }
     }
 
@@ -659,7 +664,7 @@ class Http2Stream internal constructor(
             while (sendBuffer.size > 0L) {
               emitFrame(false)
             }
-            connection.writeHeaders(id, outFinished, trailers!!.toHeaderList())
+            guardConnectionWrite { connection.writeHeaders(id, outFinished, trailers!!.toHeaderList()) }
           }
 
           hasData -> {
@@ -669,15 +674,15 @@ class Http2Stream internal constructor(
           }
 
           outFinished -> {
-            connection.writeData(id, true, null, 0L)
+            guardConnectionWrite { connection.writeData(id, true, null, 0L) }
           }
         }
       }
       withLock {
         closed = true
         notifyAll() // Because doReadTimeout() may have changed.
       }
-      connection.flush()
+      guardConnectionWrite { connection.flush() }
       cancelStreamIfNecessary()
     }
   }
@@ -717,6 +722,23 @@ class Http2Stream internal constructor(
     }
   }
 
+  /**
+   * Runs [block], which must perform a blocking write (or flush) directly against [connection]'s
+   * shared socket, bounded by [writeTimeout]. Unlike a flow-control wait, a thread inside [block]
+   * can only be freed by [StreamTimeout.timedOut] cancelling the socket -- see its doc for why.
+   */
+  @Throws(IOException::class)
+  internal fun <T> guardConnectionWrite(block: () -> T): T {
+    writeTimeout.enter()
+    try {
+      blockedInConnectionWrite = true
+      return block()
+    } finally {
+      blockedInConnectionWrite = false
+      writeTimeout.exitAndThrowIfTimedOut()
+    }
+  }
+
   /**
    * The Okio timeout watchdog will call [timedOut] if the timeout is reached. In that case we close
    * the stream (asynchronously) which will notify the waiting thread.
@@ -725,6 +747,17 @@ class Http2Stream internal constructor(
     override fun timedOut() {
       closeLater(ErrorCode.CANCEL)
       connection.sendDegradedPingLater()
+
+      // closeLater()'s notifyAll() only wakes threads parked in Object.wait() (for example a
+      // stream blocked on flow control). It can't interrupt a thread that's parked inside a
+      // blocking write to the connection's shared socket -- such as a HEADERS or DATA frame
+      // write stalled by a TCP-level network blackhole (see square/okhttp#9237). Cancelling the
+      // socket is the only way to unblock that thread. This necessarily also terminates any
+      // other streams multiplexed on the connection, but a connection that can't complete a
+      // socket write within the caller's write timeout is unresponsive for all of them anyway.
+      if (this === writeTimeout && blockedInConnectionWrite) {
+        connection.socket.cancel()
+      }
     }
 
     override fun newTimeoutException(cause: IOException?): IOException =

okhttp/src/jvmTest/kotlin/okhttp3/internal/http2/Http2ConnectionTest.kt

@@ -43,10 +43,13 @@ import okhttp3.internal.concurrent.TaskRunner
 import okhttp3.internal.concurrent.notifyAll
 import okhttp3.internal.concurrent.wait
 import okhttp3.internal.concurrent.withLock
+import okhttp3.internal.connection.BufferedSocket
 import okhttp3.internal.connection.asBufferedSocket
 import okio.AsyncTimeout
 import okio.Buffer
+import okio.BufferedSink
 import okio.BufferedSource
+import okio.Sink
 import okio.Source
 import okio.buffer
 import org.junit.jupiter.api.AfterEach
@@ -1626,6 +1629,124 @@ class Http2ConnectionTest {
     assertThat(peer.takeFrame().type).isEqualTo(Http2.TYPE_RST_STREAM)
   }
 
+  /**
+   * Regression test for https://github.com/square/okhttp/issues/9237: readTimeout/writeTimeout
+   * did nothing for an HTTP/2 connection stuck in a real TCP-level stall (e.g. a network
+   * blackhole), because [Http2Stream.StreamTimeout.timedOut] only woke threads parked in a
+   * flow-control wait -- never a thread blocked inside an actual socket write. [BlockingSocket]
+   * simulates that stall: its `sink.write()` blocks until [BlockingSocket.cancel] is called, at
+   * which point it fails, exactly like a real socket does once closed.
+   *
+   * Before the fix, this write also had no timeout wired up at all yet: [Http2Connection.newStream]
+   * wrote the initial HEADERS frame before the caller had a chance to configure the returned
+   * stream's [Http2Stream.writeTimeout]. So on a stalled connection, even a plain GET request
+   * would hang forever, regardless of any configured writeTimeout.
+   */
+  @Test fun newStreamTimesOutOnBlockedSocket() {
+    val socket = BlockingSocket()
+    val connection =
+      Http2Connection
+        .Builder(true, TaskRunner.INSTANCE)
+        .socket(socket, "blocked")
+        .build()
+    connection.start(sendConnectionPreface = false)
+
+    val startNanos = System.nanoTime()
+    assertFailsWith<InterruptedIOException> {
+      connection.newStream(headerEntries("b", "banana"), false, writeTimeoutMillis = 500L)
+    }
+    val elapsedNanos = System.nanoTime() - startNanos
+    awaitWatchdogIdle()
+
+    // Generous delta: this races the AsyncTimeout watchdog against BlockingSocket.cancel(), not
+    // just a notifyAll() wakeup like the flow-control timeout tests above.
+    assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedNanos).toDouble())
+      .isCloseTo(500.0, 1000.0)
+  }
+
+  /**
+   * Regression test for https://github.com/square/okhttp/issues/9237, covering the DATA frame
+   * write path (an in-progress request body) rather than the initial HEADERS write covered by
+   * [newStreamTimesOutOnBlockedSocket]. See that test's doc for the full explanation.
+   */
+  @Test fun dataWriteTimesOutOnBlockedSocket() {
+    val socket = BlockingSocket()
+    val connection =
+      Http2Connection
+        .Builder(true, TaskRunner.INSTANCE)
+        .socket(socket, "blocked")
+        .build()
+    connection.start(sendConnectionPreface = false)
+
+    // out=true with a fresh flow-control window: newStream() returns without touching the
+    // blocked socket because the HEADERS frame is buffered, not flushed, in that case.
+    val stream = connection.newStream(headerEntries("b", "banana"), true, writeTimeoutMillis = 0L)
+    stream.writeTimeout().timeout(500, TimeUnit.MILLISECONDS)
+
+    val startNanos = System.nanoTime()
+    assertFailsWith<InterruptedIOException> {
+      // At least EMIT_BUFFER_SIZE, so FramingSink.write() emits (and blocks on) a DATA frame
+      // without needing an explicit flush() call.
+      stream.sink.write(Buffer().write(ByteArray(16384)), 16384L)
+    }
+    val elapsedNanos = System.nanoTime() - startNanos
+    awaitWatchdogIdle()
+
+    assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedNanos).toDouble())
+      .isCloseTo(500.0, 1000.0)
+  }
+
+  /**
+   * A [BufferedSocket] that simulates a TCP-level network stall: the connection looks alive, but
+   * reads and writes block until [cancel] is called, at which point they fail -- like a real
+   * socket does once its underlying transport is closed out from under a blocked thread.
+   */
+  private class BlockingSocket : BufferedSocket {
+    private val releaseLatch = CountDownLatch(1)
+
+    override val source: BufferedSource =
+      object : Source {
+        override fun read(
+          sink: Buffer,
+          byteCount: Long,
+        ): Long = awaitRelease()
+
+        override fun timeout() = okio.Timeout.NONE
+
+        override fun close() {}
+      }.buffer()
+
+    override val sink: BufferedSink =
+      object : Sink {
+        override fun write(
+          source: Buffer,
+          byteCount: Long,
+        ) {
+          source.skip(byteCount)
+          awaitRelease()
+        }
+
+        override fun flush() {}
+
+        override fun timeout() = okio.Timeout.NONE
+
+        override fun close() {}
+      }.buffer()
+
+    private fun awaitRelease(): Nothing {
+      try {
+        releaseLatch.await()
+      } catch (e: InterruptedException) {
+        throw IOException(e)
+      }
+      throw IOException("socket cancelled")
+    }
+
+    override fun cancel() {
+      releaseLatch.countDown()
+    }
+  }
+
   @Test fun outgoingWritesAreBatched() {
     // Write the mocking script.
     peer.sendFrame().settings(Settings())