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

CHANGELOG.md

@@ -4,6 +4,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))
 - Always remove the first-draw listener after the first frame, even if no layout pass follows it ([#6152](https://github.com/getsentry/sentry-java/pull/6152))
 
 ## 8.61.1

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)
@@ -546,12 +546,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,96 @@
+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.
+ */
+internal class CapturedResponseBody(
+  private val delegate: ResponseBody,
+  private val maxBytes: Long,
+  private val onCaptured: (ByteArray) -> Unit,
+) : ResponseBody() {
+
+  private val captured = Buffer()
+  private val capturedOut = captured.outputStream()
+
+  private val reported = AtomicBoolean(false)
+
+  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() {
+    // Closing the delegate notifies listeners, which may serialize the breadcrumb this capture
+    // belongs to, so report first.
+    delegate.use { reportCaptured() }
+  }
+
+  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(capturedOut, sinkBefore, toTake)
+            }
+            captured.size >= maxBytes
+          }
+        if (capFull) {
+          reportCaptured()
+        }
+      } else if (read == -1L) {
+        reportCaptured()
+      }
+      return read
+    }
+
+    override fun close() {
+      try {
+        reportCaptured()
+      } finally {
+        super.close()
+      }
+    }
+  }
+
+  private fun reportCaptured() {
+    if (reported.compareAndSet(false, true)) {
+      onCaptured(synchronized(captured) { captured.clone().readByteArray() })
+    }
+  }
+}

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

@@ -20,7 +20,6 @@ import io.sentry.TypeCheckHint.OKHTTP_REQUEST
 import io.sentry.TypeCheckHint.OKHTTP_RESPONSE
 import io.sentry.TypeCheckHint.SENTRY_REPLAY_NETWORK_DETAILS
 import io.sentry.clientreport.DiscardReason
-import io.sentry.okhttp.SentryOkHttpInterceptor.BeforeSpanCallback
 import io.sentry.transport.CurrentDateProvider
 import io.sentry.util.ExceptionUtils
 import io.sentry.util.IntegrationUtils.addIntegrationToSdkVersion
@@ -54,6 +53,7 @@ import org.jetbrains.annotations.VisibleForTesting
  * @param failedRequestTargets The SDK will only capture HTTP Client errors if the HTTP Request URL
  *   is a match for any of the defined targets.
  */
+@Suppress("TooManyFunctions") // one function per concern of the interception pipeline
 public open class SentryOkHttpInterceptor(
   private val scopes: IScopes = ScopesAdapter.getInstance(),
   private val beforeSpan: BeforeSpanCallback? = null,
@@ -195,6 +195,9 @@ public open class SentryOkHttpInterceptor(
         }
       }
 
+      // after the client error was captured from the untouched response
+      response = response.withNetworkBodyCapture(networkDetailData)
+
       return response
     } catch (e: IOException) {
       span?.apply {
@@ -207,20 +210,6 @@ public open class SentryOkHttpInterceptor(
       // this only works correctly if SentryOkHttpInterceptor is the last one in the chain
       okHttpEvent?.setRequest(request)
 
-      response?.let {
-        networkDetailData?.setResponseDetails(
-          it.code,
-          NetworkDetailCaptureUtils.createResponse(
-            it,
-            it.body?.contentLength(),
-            scopes.options.sessionReplay.isNetworkCaptureBodies,
-            { resp: Response -> resp.extractResponseBody(scopes.options.logger) },
-            scopes.options.sessionReplay.networkResponseHeaders,
-            { resp: Response -> resp.headers.toMap() },
-          ),
-        )
-      }
-
       // Set network details on the OkHttpEvent so it can include them in the breadcrumb hint
       okHttpEvent?.setNetworkDetails(networkDetailData)
 
@@ -321,41 +310,95 @@ public open class SentryOkHttpInterceptor(
         logger,
       )
     } catch (e: Exception) {
-      logger.log(io.sentry.SentryLevel.ERROR, "Failed to parse buffered request body: ${e.message}")
+      logger.log(SentryLevel.ERROR, "Failed to parse buffered request body: ${e.message}")
       return null
     }
   }
 
-  /** Extracts the body content from an OkHttp Response safely */
-  private fun Response.extractResponseBody(logger: ILogger): NetworkBody? {
-    return body?.let { responseBody ->
-      try {
-        val contentType = responseBody.contentType()
-        val contentTypeString = contentType?.toString()
-        val maxBodySize = SentryReplayOptions.MAX_NETWORK_BODY_SIZE
-
-        // Peek at the body (doesn't consume it)
-        // We +1 here in order to properly truncate within NetworkBodyParser.fromBytes
-        // and be able to distinguish from an oversized request and a request matching maxBodySize
-        val peekBody = peekBody(maxBodySize.toLong() + 1)
-        val bodyBytes = peekBody.bytes()
-
-        val charset = contentType?.charset(Charsets.UTF_8)?.name() ?: "UTF-8"
-        return NetworkBodyParser.fromBytes(
-          bodyBytes,
-          contentTypeString,
-          charset,
-          maxBodySize,
-          logger,
-        )
-      } catch (e: Exception) {
-        logger.log(
-          io.sentry.SentryLevel.ERROR,
-          "Failed to read http response body for Network Details: ${e.message}",
-        )
-        null
-      }
+  /**
+   * Returns this response with a body whose content is captured for replay while the application
+   * reads it.
+   *
+   * Reading the body up front, as [Response.peekBody] does, deadlocks on a response of unknown
+   * length: server-sent events, a long-poll, a chunked endpoint that stays open, and every response
+   * OkHttp decompressed itself. The caller would never receive the response at all.
+   *
+   * Every body therefore goes through [CapturedResponseBody], which captures only what the
+   * application reads. A body nobody reads leaves the status code and the headers recorded below,
+   * and no body.
+   */
+  private fun Response.withNetworkBodyCapture(networkDetailData: NetworkRequestData?): Response {
+    val responseBody = body
+    if (networkDetailData == null || responseBody == null) {
+      return this
+    }
+
+    val knownBodySize = responseBody.contentLength().takeIf { it >= 0L }
+
+    // The status code and the headers are known now, so record them before anything is consumed. A
+    // stream that stays open for minutes never gets further than this.
+    recordResponse(networkDetailData, knownBodySize) { null }
+
+    if (!scopes.options.sessionReplay.isNetworkCaptureBodies) {
+      return this
     }
+
+    val logger = scopes.options.logger
+    // One byte above the limit, so NetworkBodyParser.fromBytes can tell a truncated body from one
+    // that happens to match the limit exactly.
+    val captureCap = SentryReplayOptions.MAX_NETWORK_BODY_SIZE.toLong() + 1
+    val contentType = responseBody.contentType()
+    val contentTypeString = contentType?.toString()
+    val charset = contentType?.charset(Charsets.UTF_8)?.name() ?: "UTF-8"
+
+    return newBuilder()
+      .body(
+        CapturedResponseBody(responseBody, captureCap) { capturedBytes ->
+          // This runs on the thread consuming the body, inside its read or close, so a failure in
+          // here must stay in here.
+          try {
+            recordResponse(networkDetailData, knownBodySize ?: capturedBytes.size.toLong()) {
+              NetworkBodyParser.fromBytes(
+                capturedBytes,
+                contentTypeString,
+                charset,
+                SentryReplayOptions.MAX_NETWORK_BODY_SIZE,
+                logger,
+              )
+            }
+          } catch (@Suppress("TooGenericExceptionCaught") e: Throwable) {
+            ExceptionUtils.rethrowIfFatal(e)
+            logger.log(
+              SentryLevel.ERROR,
+              "Failed to capture the http response body for Network Details",
+              e,
+            )
+          }
+        }
+      )
+      .build()
+  }
+
+  /**
+   * Records what is known about this response, replacing anything recorded for it before. Called
+   * once before the body is consumed and once with the captured body.
+   */
+  private fun Response.recordResponse(
+    networkDetailData: NetworkRequestData,
+    bodySize: Long?,
+    body: () -> NetworkBody?,
+  ) {
+    networkDetailData.setResponseDetails(
+      NetworkDetailCaptureUtils.createResponse(
+        this,
+        code,
+        bodySize,
+        true,
+        { body() },
+        scopes.options.sessionReplay.networkResponseHeaders,
+        { resp: Response -> resp.headers.toMap() },
+      )
+    )
   }
 
   private fun finishSpan(