Repository navigation
fix(okhttp): Don't block on response bodies by reading the body upfront #6231
+1,207
−77
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| 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, 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 | ||
|
Member
There was a problem hiding this comment. l: Double-checking that the |
||
| * 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) | ||
|
|
||
| // A buffered source still forwards a short event on its own: one read takes at most one segment | ||
| // from the source below and returns with whatever arrived. 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 { | ||
|
Member
There was a problem hiding this comment. nit: we can replace |
||
| // 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. | ||
|
Member
There was a problem hiding this comment. nit: Fwiw, I can't quite figure out what the LLM was trying to say here. |
||
| 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) | ||
|
Member
There was a problem hiding this comment.
Looks like
My LLM tells me that for SSE, long-poll, or chunked streams with small events, each small event will be forced to retain a full 8 KB array. With 10-byte heartbeats, that makes for ~120 MB of memory leaked per stream. The Buffer.copyTo() method that takes an OutputStream doesn't share the segment, and so we avoid leaking it (link). There was a problem hiding this comment. Shared segments inflate capture memoryHigh Severity
Additional Locations (1)  Reviewed by Cursor Bugbot for commit 4dbcbfa. Configure here. |
||
| } | ||
| 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() }) | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -54,6 +54,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 +196,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 +211,6 @@ public open class SentryOkHttpInterceptor( | |
| // this only works correctly if SentryOkHttpInterceptor is the last one in the chain | ||
| okHttpEvent?.setRequest(request) | ||
|
|
||
| // Set network details on the OkHttpEvent so it can include them in the breadcrumb hint | ||
| okHttpEvent?.setNetworkDetails(networkDetailData) | ||
|
|
||
|
|
@@ -326,36 +316,89 @@ public open class SentryOkHttpInterceptor( | |
| } | ||
| } | ||
|
|
||
| /** | ||
| * 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 | ||
|
Member
There was a problem hiding this comment. h: Thoughts about calling I'm worried about delaying delivery of SSE events when the |
||
| // 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( | ||
| io.sentry.SentryLevel.ERROR, | ||
| "Failed to capture the http response body for Network Details: ${e.message}", | ||
|
Member
There was a problem hiding this comment. l: We could also log |
||
| ) | ||
| } | ||
|
There was a problem hiding this comment. New broad Throwable catch addedLow Severity This new Triggered by project rule: PR Review Guidelines for Cursor Bugbot Reviewed by Cursor Bugbot for commit 4dbcbfa. Configure here. |
||
| } | ||
| ) | ||
| .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( | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
l: Might want to add a "Session Replay:" prefix to this so we don't suggest this was happening to everyone (could also add a qualification that it only impacted folks who had network details turned on if you think that'd be helpful).