fix(okhttp): Don't block on response bodies by reading the body upfront

CHANGELOG.md

@@ -15,6 +15,7 @@
 
 ### Fixes
 
+- Session Replay: Fix `SentryOkHttpInterceptor` hanging forever on responses whose body has no known length, such as Server-Sent Events or a gzipped response, when network details are enabled for the URL ([#6231](https://github.com/getsentry/sentry-java/pull/6231))
 - Stop marking an OkHttp call span as failed when a connection attempt to one of the host's addresses fails but OkHttp connects on another ([#6472](https://github.com/getsentry/sentry-java/pull/6472))
 
 ### Dependencies

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,12 @@ public open class DefaultReplayBreadcrumbConverter() : ReplayBreadcrumbConverter
 
       // Add Network Details data when available
       networkDetailData?.let { networkData ->
+        val responseDetails = networkData.response
+
         networkData.method?.let { breadcrumbData["method"] = it }
-        networkData.statusCode?.let { breadcrumbData["statusCode"] = it }
+        responseDetails?.statusCode?.let { breadcrumbData["statusCode"] = it }
         networkData.requestBodySize?.let { breadcrumbData["requestBodySize"] = it }
-        networkData.responseBodySize?.let { breadcrumbData["responseBodySize"] = it }
+        responseDetails?.size?.let { breadcrumbData["responseBodySize"] = it }
 
         networkData.request?.let { request ->
           val requestData = mutableMapOf<String, Any?>()
@@ -257,7 +259,7 @@ public open class DefaultReplayBreadcrumbConverter() : ReplayBreadcrumbConverter
           }
         }
 
-        networkData.response?.let { response ->
+        responseDetails?.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,12 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     fakeOkHttpNetworkDetails.setResponseDetails(
-      200,
       ReplayNetworkRequestOrResponse(
+        200,
         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 +491,12 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     fakeOkHttpNetworkDetails.setResponseDetails(
-      404,
       ReplayNetworkRequestOrResponse(
+        404,
         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)
@@ -528,6 +528,47 @@ class DefaultReplayBreadcrumbConverterTest {
     assertEquals(mapOf("Content-Type" to "text/plain"), responseData["headers"])
   }
 
+  @Test
+  fun `a response body captured after the breadcrumb was converted does not reach the replay`() {
+    // Accepted gap: a segment can be cut between the call returning and the app reading the body.
+    val options = SentryOptions.empty()
+    val converter = fixture.getSut(options)
+    val httpBreadcrumb =
+      Breadcrumb(Date(123L)).apply {
+        type = "http"
+        category = "http"
+        data["url"] = "https://example.com"
+        data[SpanDataConvention.HTTP_START_TIMESTAMP] = 1000L
+        data[SpanDataConvention.HTTP_END_TIMESTAMP] = 2000L
+      }
+    val networkRequestData = NetworkRequestData("GET")
+    networkRequestData.setResponseDetails(
+      ReplayNetworkRequestOrResponse(200, null, null, mapOf("Content-Type" to "application/json"))
+    )
+    // Simulates Scopes.addBreadcrumb, which hands the network details to the converter
+    options.beforeBreadcrumb?.execute(
+      httpBreadcrumb,
+      Hint().apply { set(SENTRY_REPLAY_NETWORK_DETAILS, networkRequestData) },
+    )
+
+    val rrwebEvent = converter.convert(httpBreadcrumb)
+    networkRequestData.setResponseDetails(
+      ReplayNetworkRequestOrResponse(
+        200,
+        21L,
+        NetworkBody("response body content"),
+        mapOf("Content-Type" to "application/json"),
+      )
+    )
+
+    check(rrwebEvent is RRWebSpanEvent)
+    assertEquals(200, rrwebEvent.data!!["statusCode"])
+    val responseData = rrwebEvent.data!!["response"] as Map<*, *>
+    assertEquals(mapOf("Content-Type" to "application/json"), responseData["headers"])
+    assertNull(responseData["body"])
+    assertNull(rrwebEvent.data!!["responseBodySize"])
+  }
+
   @Test
   fun `does not convert network details data for non-http breadcrumbs`() {
     val navigationBreadcrumb =
@@ -546,12 +587,12 @@ class DefaultReplayBreadcrumbConverterTest {
       )
     )
     networkRequestData.setResponseDetails(
-      200,
       ReplayNetworkRequestOrResponse(
+        200,
         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,116 @@
+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.
+ *
+ * @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, as soon as the capture can no longer grow: the cap is
+ *   reached, the stream ends, a read fails, or the body is closed. Runs synchronously on the thread
+ *   that read or closed the body, so it must not block. A body abandoned without any of those never
+ *   reaches it. `isComplete` is false when the bytes are only a prefix of the body.
+ */
+internal class CapturedResponseBody(
+  private val delegate: ResponseBody,
+  private val maxBytes: Long,
+  private val onCaptured: (bytes: ByteArray, isComplete: Boolean) -> Unit,
+) : ResponseBody() {
+
+  private val captured = Buffer()
+  private val capturedOut = captured.outputStream()
+
+  private val reported = AtomicBoolean(false)
+  @Volatile private var bytesRead = 0L
+  @Volatile private var endOfStream = false
+
+  private val capturingSourceLazy = lazy { CapturingSource(delegate.source()).buffer() }
+  private val capturingSource: BufferedSource by capturingSourceLazy
+
+  override fun contentType(): MediaType? = delegate.contentType()
+
+  override fun contentLength(): Long = delegate.contentLength()
+
+  override fun source(): BufferedSource = capturingSource
+
+  override fun close() {
+    // Closing the delegate notifies listeners, which may serialize the breadcrumb this capture
+    // belongs to, so report first.
+    delegate.use {
+      try {
+        reportCaptured()
+      } finally {
+        // the application may still hold the source, and must not read buffered bytes from it
+        if (capturingSourceLazy.isInitialized()) {
+          capturingSource.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) {
+        bytesRead += read
+        if (!reported.get()) {
+          val capFull =
+            synchronized(captured) {
+              // checked again under the lock, so nothing is copied into a drained capture
+              if (reported.get()) {
+                return@synchronized false
+              }
+              // 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(capturedOut, sinkBefore, toTake)
+              }
+              captured.size >= maxBytes
+            }
+          if (capFull) {
+            reportCaptured()
+          }
+        }
+      } else if (read == -1L) {
+        endOfStream = true
+        reportCaptured()
+      }
+      return read
+    }
+
+    override fun close() {
+      try {
+        reportCaptured()
+      } finally {
+        super.close()
+      }
+    }
+  }
+
+  private fun reportCaptured() {
+    if (reported.compareAndSet(false, true)) {
+      val length = delegate.contentLength()
+      val isComplete = endOfStream || (length >= 0L && bytesRead >= length)
+      onCaptured(synchronized(captured) { captured.readByteArray() }, isComplete)
+    }
+  }
+}