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/CapturedResponseBody.kt

@@ -0,0 +1,114 @@
+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 reads into a capped buffer.
+ *
+ * Reading the body up front instead, with [okhttp3.Response.peekBody], deadlocks on a response of
+ * unknown length: a peek is `request(byteCount)`, which waits for that many bytes or for the end of
+ * the stream, and a server-sent-events stream, a long-poll or an open chunked endpoint delivers
+ * neither. The caller would never receive the response at all.
+ *
+ * Nothing is therefore read on the capture's own account. Only what the application reads is
+ * captured, which also keeps the cost of the instrumentation to a copy of those bytes.
+ *
+ * @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 — so
+ *   it must not block. A body that is abandoned without being read to the end or closed never
+ *   reaches it.
+ */
+internal class CapturedResponseBody(
+  private val delegate: ResponseBody,
+  private val maxBytes: Long,
+  private val onCaptured: (ByteArray) -> Unit,
+) : ResponseBody() {
+
+  private val captured = Buffer()
+  private val reported = AtomicBoolean(false)
+
+  // Buffering the capturing source cannot re-introduce the deadlock above: a BufferedSource read
+  // takes at most one segment from the source below it and returns with whatever arrived, so a
+  // short event is still forwarded on its own. Only request/require/readByteArray() wait for a byte
+  // count, and those are the application's own calls.
+  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 {
+      // Closing the delegate notifies listeners, which may serialize the breadcrumb this capture
+      // belongs to, so report first.
+      reportCaptured()
+    } finally {
+      delegate.close()
+    }
+  }
+
+  private inner class CapturingSource(source: Source) : ForwardingSource(source) {
+    override fun read(sink: Buffer, byteCount: Long): Long {
+      val sinkBefore = sink.size
+      val read =
+        try {
+          super.read(sink, byteCount)
+        } catch (e: IOException) {
+          // What arrived is the evidence for this very failure, and a caller handling it is not
+          // obliged to close the body.
+          reportCaptured()
+          throw e
+        }
+
+      if (read > 0L) {
+        val capFull =
+          synchronized(captured) {
+            // the application may not have taken everything in the sink yet, hence the offset
+            val toTake = minOf(maxBytes - captured.size, sink.size - sinkBefore)
+            if (toTake > 0L) {
+              sink.copyTo(captured, sinkBefore, toTake)
+            }
+            captured.size >= maxBytes
+          }
+        if (capFull) {
+          reportCaptured()
+        }
+      } else if (read == -1L) {
+        reportCaptured()
+      }
+      return read
+    }
+
+    override fun close() {
+      try {
+        reportCaptured()
+      } finally {
+        super.close()
+      }
+    }
+  }
+
+  /**
+   * [captured] is written by the thread reading the body and read here, which [close] may reach
+   * from another thread — cancelling a stream from elsewhere is ordinary use — and [Buffer] is not
+   * thread-safe.
+   */
+  private fun reportCaptured() {
+    if (reported.compareAndSet(false, true)) {
+      onCaptured(synchronized(captured) { captured.clone().readByteArray() })
+    }
+  }
+}