Repository navigation
fix(okhttp): Don't block on response bodies by reading the body upfront #6231
+1,203
−79
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,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() } | ||
| } | ||
|
There was a problem hiding this comment. Body close leaves source readableLow Severity
  Reviewed by Cursor Bugbot for commit da019c0. Configure here. |
||
|
|
||
| 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. | ||
|
markushi marked this conversation as resolved.
|
||
| 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() }) | ||
|
Member
There was a problem hiding this comment. m: Thoughts about draining instead of clone here + skip copying once reported? Otherwise a long SSE stream might keep the buffer alive and acquire the lock on every read. |
||
| } | ||
| } | ||
| } | ||
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 |
|---|---|---|
|
|
@@ -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.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) | ||
|
|
||
| // 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(SentryLevel.ERROR, "Failed to parse buffered request body: ${e.message}") | ||
| return 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 | ||
|
0xadam-brown marked this conversation as resolved.
|
||
| // here must stay in here. | ||
| try { | ||
| recordResponse(networkDetailData, knownBodySize ?: capturedBytes.size.toLong()) { | ||
|
Member
There was a problem hiding this comment. l: When a body is closed early or a read fails, is the prefix treated as a complete body? Wondering whether a cut-off JSON body would show up as INVALID_JSON rather than truncated, and whether the size would be off... |
||
| NetworkBodyParser.fromBytes( | ||
| capturedBytes, | ||
| contentTypeString, | ||
| charset, | ||
| SentryReplayOptions.MAX_NETWORK_BODY_SIZE, | ||
| logger, | ||
| ) | ||
| } | ||
|
There was a problem hiding this comment. Replay drops late-captured response bodyMedium Severity The interceptor now records status and headers immediately, then replaces them with the body later via Additional Locations (1)Reviewed by Cursor Bugbot for commit da019c0. Configure here. |
||
| } catch (@Suppress("TooGenericExceptionCaught") e: Throwable) { | ||
| ExceptionUtils.rethrowIfFatal(e) | ||
| logger.log( | ||
| SentryLevel.ERROR, | ||
| "Failed to capture the http response body for Network Details", | ||
| e, | ||
| ) | ||
| } | ||
|
markushi marked this conversation as resolved.
|
||
| } | ||
| ) | ||
| .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.
m: Should close() also close capturingSource if it was created? Without it, seems the app could keep reading buffered bytes.