fix(okhttp): Don't block on response bodies with no known length

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Fixes
+
+- Fix `SentryOkHttpInterceptor` hanging forever on responses whose body has no known length, such as Server-Sent Events or a gzipped response ([#6231](https://github.com/getsentry/sentry-java/pull/6231))
+
 ### Features
 
 - Report the cellular network technology generation in `device.connection_effective_type`, for example `4g` or `5g` ([#6146](https://github.com/getsentry/sentry-java/pull/6146))

sentry-android-integration-tests/test-app-size/proguard-rules.pro

@@ -47,5 +47,7 @@
 -dontwarn kotlin.math.MathKt
 -dontwarn okhttp3.EventListener
 -dontwarn okhttp3.Interceptor
+-dontwarn okhttp3.ResponseBody
+-dontwarn okio.ForwardingSource
 # Assume all classes are used to not strip them out, e.g. integrations like Compose or Sqlite
 -keep class io.sentry.**

sentry-android-replay/src/main/java/io/sentry/android/replay/DefaultReplayBreadcrumbConverter.kt

@@ -233,10 +233,14 @@ public open class DefaultReplayBreadcrumbConverter() : ReplayBreadcrumbConverter
 
       // Add Network Details data when available
       networkDetailData?.let { networkData ->
+        // One snapshot of the response: an integration that captures a streamed body may fill it in
+        // while this runs, and the status code must then belong to the body next to it.
+        val responseDetails = networkData.responseDetails
+
         networkData.method?.let { breadcrumbData["method"] = it }
-        networkData.statusCode?.let { breadcrumbData["statusCode"] = it }
+        responseDetails?.let { breadcrumbData["statusCode"] = it.statusCode }
         networkData.requestBodySize?.let { breadcrumbData["requestBodySize"] = it }
-        networkData.responseBodySize?.let { breadcrumbData["responseBodySize"] = it }
+        responseDetails?.response?.size?.let { breadcrumbData["responseBodySize"] = it }
 
         networkData.request?.let { request ->
           val requestData = mutableMapOf<String, Any?>()
@@ -257,7 +261,7 @@ public open class DefaultReplayBreadcrumbConverter() : ReplayBreadcrumbConverter
           }
         }
 
-        networkData.response?.let { response ->
+        responseDetails?.response?.let { response ->
           val responseData = mutableMapOf<String, Any?>()
           response.size?.let { responseData["size"] = it }
           response.body?.let {

sentry-android-replay/src/test/java/io/sentry/android/replay/DefaultReplayBreadcrumbConverterTest.kt

@@ -429,12 +429,14 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     fakeOkHttpNetworkDetails.setResponseDetails(
-      200,
-      ReplayNetworkRequestOrResponse(
-        500L,
-        NetworkBody(mapOf("status" to "success", "message" to "OK")),
-        mapOf("Content-Type" to "text/plain"),
-      ),
+      NetworkRequestData.ResponseDetails(
+        200,
+        ReplayNetworkRequestOrResponse(
+          500L,
+          NetworkBody(mapOf("status" to "success", "message" to "OK")),
+          mapOf("Content-Type" to "text/plain"),
+        ),
+      )
     )
     val hintWithFakeOKHttpNetworkDetails = Hint()
     hintWithFakeOKHttpNetworkDetails.set(SENTRY_REPLAY_NETWORK_DETAILS, fakeOkHttpNetworkDetails)
@@ -491,12 +493,14 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     fakeOkHttpNetworkDetails.setResponseDetails(
-      404,
-      ReplayNetworkRequestOrResponse(
-        550L,
-        NetworkBody(mapOf("status" to "success", "message" to "OK")),
-        mapOf("Content-Type" to "text/plain"),
-      ),
+      NetworkRequestData.ResponseDetails(
+        404,
+        ReplayNetworkRequestOrResponse(
+          550L,
+          NetworkBody(mapOf("status" to "success", "message" to "OK")),
+          mapOf("Content-Type" to "text/plain"),
+        ),
+      )
     )
     val hintWithFakeOKHttpNetworkDetails = Hint()
     hintWithFakeOKHttpNetworkDetails.set(SENTRY_REPLAY_NETWORK_DETAILS, fakeOkHttpNetworkDetails)
@@ -546,12 +550,14 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     networkRequestData.setResponseDetails(
-      200,
-      ReplayNetworkRequestOrResponse(
-        100L,
-        NetworkBody("response body content"),
-        mapOf("Content-Type" to "application/json"),
-      ),
+      NetworkRequestData.ResponseDetails(
+        200,
+        ReplayNetworkRequestOrResponse(
+          100L,
+          NetworkBody("response body content"),
+          mapOf("Content-Type" to "application/json"),
+        ),
+      )
     )
     hint.set(SENTRY_REPLAY_NETWORK_DETAILS, networkRequestData)
 

sentry-okhttp/src/main/java/io/sentry/okhttp/NetworkBodyCapturingResponseBody.kt

@@ -0,0 +1,145 @@
+package io.sentry.okhttp
+
+import java.io.IOException
+import java.util.concurrent.atomic.AtomicBoolean
+import okhttp3.MediaType
+import okhttp3.ResponseBody
+import okio.Buffer
+import okio.BufferedSource
+import okio.ForwardingSource
+import okio.Source
+import okio.buffer
+
+/**
+ * A [ResponseBody] that copies the bytes the application consumes into a capped buffer.
+ *
+ * Capturing a body by peeking at it up front does not work for every response. OkHttp implements
+ * [okhttp3.Response.peekBody] as `request(byteCount)`, which keeps reading until the requested
+ * number of bytes is buffered or the stream ends. A response of unknown length may never end — a
+ * server-sent-events stream, a long-poll, a chunked endpoint that stays open — so the peek never
+ * returns and the thread that called `execute()` never gets the response at all.
+ *
+ * This body instead captures what is actually consumed. It forwards every byte to the application
+ * untouched and hands the captured bytes to [onCaptured] as soon as the capture can no longer grow,
+ * which is when the stream ends or breaks, when the application closes the body, or when the cap is
+ * reached.
+ *
+ * A body the application never reads is still captured if its length is known: such a body ends on
+ * its own, so taking it while closing is bounded. That is how the body of a response whose status
+ * code was the only thing of interest reaches the capture.
+ *
+ * @param delegate the body to capture from.
+ * @param maxBytes the maximum number of bytes to retain; capture stops once it is reached.
+ * @param onCaptured invoked at most once, synchronously, on the thread that finishes the body. It
+ *   is never invoked for a body that is abandoned without being read to the end or closed.
+ */
+internal class NetworkBodyCapturingResponseBody(
+  private val delegate: ResponseBody,
+  private val maxBytes: Long,
+  private val onCaptured: (ByteArray) -> Unit,
+) : ResponseBody() {
+
+  private val captured = Buffer()
+  private val reported = AtomicBoolean(false)
+  @Volatile private var readStarted = false
+
+  // A ResponseBody must hand out a BufferedSource, so the capturing source is buffered. That cannot
+  // re-introduce the blocking this class exists to avoid: BufferedSource.read(sink, byteCount)
+  // issues at most one segment-sized read on the source below it and returns with whatever arrived,
+  // so a short event is still forwarded on its own. The reads that loop until a byte count is
+  // reached — request, require, readByteArray() — are the application's own choice, and the capture
+  // never calls them.
+  private val capturingSource: BufferedSource by lazy {
+    CapturingSource(delegate.source()).buffer()
+  }
+
+  override fun contentType(): MediaType? = delegate.contentType()
+
+  override fun contentLength(): Long = delegate.contentLength()
+
+  override fun source(): BufferedSource = capturingSource
+
+  override fun close() {
+    try {
+      captureUnreadBody()
+      // Report before closing the delegate: closing it notifies listeners, which may serialize the
+      // breadcrumb this capture belongs to.
+      reportCaptured()
+    } finally {
+      delegate.close()
+    }
+  }
+
+  /**
+   * Reads a body nobody touched, so it is captured as well. Only for a body of known length, which
+   * ends on its own, and only while no one is reading: a concurrent read of the same body is what
+   * OkHttp forbids.
+   */
+  @Suppress("SwallowedException") // the body is being closed; a failure here leaves it as it was
+  private fun captureUnreadBody() {
+    if (readStarted || reported.get() || delegate.contentLength() < 0L) {
+      return
+    }
+    try {
+      capturingSource.request(maxBytes)
+    } catch (e: IOException) {
+      // nothing more to capture
+    }
+  }
+
+  /** Copies the bytes passing through into [captured]. */
+  private inner class CapturingSource(source: Source) : ForwardingSource(source) {
+    override fun read(sink: Buffer, byteCount: Long): Long {
+      readStarted = true
+      val sinkBefore = sink.size
+      val read =
+        try {
+          super.read(sink, byteCount)
+        } catch (e: IOException) {
+          // The stream broke, so nothing more can arrive. What did arrive is the evidence for this
+          // very failure, and a caller handling the error is not obliged to close the body.
+          reportCaptured()
+          throw e
+        }
+
+      if (read > 0L) {
+        val capFull =
+          synchronized(captured) {
+            val toTake = minOf(maxBytes - captured.size, sink.size - sinkBefore)
+            if (toTake > 0L) {
+              sink.copyTo(captured, sinkBefore, toTake)
+            }
+            captured.size >= maxBytes
+          }
+        if (capFull) {
+          // the capture can never grow again, so this is the moment it becomes final
+          reportCaptured()
+        }
+      } else if (read == -1L) {
+        // end of stream: nothing more can ever arrive
+        reportCaptured()
+      }
+      return read
+    }
+
+    override fun close() {
+      try {
+        reportCaptured()
+      } finally {
+        super.close()
+      }
+    }
+  }
+
+  /**
+   * The consuming thread writes [captured] from [CapturingSource.read] while [close] may be called
+   * by another thread — cancelling a stream from elsewhere is ordinary use — and [Buffer] is not
+   * thread-safe, so both accesses are guarded. [reported] keeps the callback to a single
+   * invocation.
+   */
+  private fun reportCaptured() {
+    if (reported.compareAndSet(false, true)) {
+      onCaptured(synchronized(captured) { captured.clone().readByteArray() })
+    }
+  }
+}