fix(replay): Prevent concurrent PixelCopy frame access (#5808)

* fix(replay): Prevent concurrent PixelCopy frame access

Keep PixelCopy, masking, compositing, and cleanup from accessing the shared bitmap concurrently.

Fixes GH-5340

Co-Authored-By: Codex <noreply@openai.com>

* changelog

* fix(android-replay): Preserve pending PixelCopy capture

Re-arm the recorder's content-change gate when a capture is skipped because another frame is still in flight. This ensures the latest UI state is retried on the next capture interval.

Refs GH-5340
Co-Authored-By: OpenAI Codex <noreply@openai.com>

* fix(replay): Release PixelCopy frame gate and cleanup on failure paths

Two bugbot findings on #5808:

- Frame gate stuck on callback errors: viewhierarchy traversal or
  captureSurfaceViews throwing between the PixelCopy success check and
  the executor submit left frameInFlight = true forever, silently
  wedging all future captures. Wrap the post-success block in a
  try/finally that releases the gate unless work was successfully
  handed off.
- Cleanup lost after executor shutdown: close() typically runs after
  ReplayIntegration has shut down the replay executor, so submit()
  returns null and the bitmap + maskRenderer were never released. Fall
  back to running cleanup inline in that case.

* chore(replay): Drop redundant handedOff flag in capture callback

The mask branch already knows via 'submitted == null' whether the executor
took ownership; the surface-view branch always hands off. Flatten to
explicit finishFrame() calls on the two failure paths (mask null-submit,
outer catch) instead of tracking handoff state across a try/finally.

* fix(replay): Distinguish inline execution from executor rejection

ReplayExecutorService.submit previously returned null both when the caller
was on the worker thread (task ran inline) and when the executor rejected
the submission (task did NOT run). Callers had no way to tell them apart.

Return a CompletedFuture sentinel for inline execution; null now means
only rejection. Also narrow PixelCopyStrategy's frame-processing catch
from Throwable to RuntimeException so OOM/LinkageError still propagate.

* chore(replay): Drop redundant cleanupScheduled guard

Cleanup body is already idempotent (screenshot.isRecycled check,
MaskRenderer.close guards on isInitialized + isRecycled). A stray extra
scheduleCleanup would just submit a no-op — not worth the AtomicBoolean.

* test(replay): Fix recursive close and assert masking is skipped on close race

The mock executor closed the strategy on every submit(); since close()
itself submits the cleanup task, this recursed
close() -> scheduleCleanup() -> submit() until the stack overflowed.
Close only when the mask task is submitted.

The prior "does not crash" assertion was also vacuous under the
frameInFlight gate (passed even with the isClosed guard removed). Assert
instead that no screenshot is emitted once close() races masking, which
fails if the guard in applyMaskingAndNotify is removed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(replay): Guard finishFrame cleanup with CAS to avoid recycling a bitmap a new capture is using

finishFrame cleared frameInFlight before checking isClosed, so a new capture could
take the gate and start PixelCopy into the shared screenshot while the old
finishFrame went on to scheduleCleanup after close() flipped isClosed — recycling
the bitmap mid-write. Re-take the gate with compareAndSet before cleaning up so the
losing frame backs off.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(replay): Make close's idle cleanup claim the gate atomically

close() checked !frameInFlight.get() non-atomically before scheduleCleanup, so a
capture racing in after the check could take the gate, see isClosed, and have its
finishFrame schedule a second cleanup. Extract the shared "claim the gate, then the
winner cleans up once" invariant into cleanUpIfIdle() so close() and finishFrame()
use the same CAS.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(replay): Return CompletedFuture from inlineExecutor to match ReplayExecutorService

inlineExecutor() returned null from submit, which under the executor's contract means
"rejected" — making capture()'s null-fallback finishFrame run on top of the mask
task's own finally, a double-release production never hits on the inline path. Return
CompletedFuture so the fixture matches the real inline semantics.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(replay): Hold frame gate across emitLastScreenshot to prevent concurrent bitmap access

emitLastScreenshot runs on the main thread, so the downstream consumer queues
its bitmap read (JPEG compress) to the executor asynchronously. Without the
gate, the next capture tick's PixelCopy.request writes into the shared
screenshot while the queued read is still in flight. Submit the consumer call
to the executor so it runs inline on the worker thread while the gate is held,
same pattern as the masked capture path.

Also fixes the close-race test mock to return CompletedFuture instead of null,
matching the executor contract (same fix as inlineExecutor in cd3a7c0).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Codex <noreply@openai.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Fixes
 
+- Prevent concurrent PixelCopy access during Session Replay masking and bitmap cleanup ([#5808](https://github.com/getsentry/sentry-java/pull/5808))
 - Release `MediaMuxer` when the replay video encoder fails to start to avoid a resource leak ([#5607](https://github.com/getsentry/sentry-java/pull/5607))
 
 ### Performance

sentry-android-replay/src/main/java/io/sentry/android/replay/screenshot/PixelCopyStrategy.kt

@@ -59,6 +59,7 @@ internal class PixelCopyStrategy(
   private val contentChanged = AtomicBoolean(false)
   private val unstableCaptures = AtomicInteger(0)
   private val isClosed = AtomicBoolean(false)
+  private val frameInFlight = AtomicBoolean(false)
   private val dstOverPaint by
     lazy(NONE) { Paint().apply { xfermode = PorterDuffXfermode(PorterDuff.Mode.DST_OVER) } }
   private val screenshotCanvas by lazy(NONE) { Canvas(screenshot) }
@@ -77,8 +78,15 @@ internal class PixelCopyStrategy(
       return
     }
 
+    if (!frameInFlight.compareAndSet(false, true)) {
+      options.logger.log(DEBUG, "PixelCopyStrategy capture is already in flight, skipping")
+      markContentChanged()
+      return
+    }
+
     if (isClosed.get()) {
       options.logger.log(DEBUG, "PixelCopyStrategy is closed, not capturing screenshot")
+      finishFrame()
       return
     }
 
@@ -90,51 +98,72 @@ internal class PixelCopyStrategy(
         { copyResult: Int ->
           if (isClosed.get()) {
             options.logger.log(DEBUG, "PixelCopyStrategy is closed, ignoring capture result")
+            finishFrame()
             return@request
           }
 
           if (copyResult != PixelCopy.SUCCESS) {
             options.logger.log(INFO, "Failed to capture replay recording: %d", copyResult)
             unstableCaptures.set(0)
             lastCaptureSuccessful.set(false)
+            finishFrame()
             return@request
           }
 
           val changedDuringCapture = contentChanged.get()
           if (changedDuringCapture && shouldSkipUnstableCapture()) {
+            finishFrame()
             return@request
           }
 
-          // TODO: disableAllMasking here and dont traverse?
-          val viewHierarchy = ViewHierarchyNode.fromView(root, null, 0, options.sessionReplay)
-          val surfaceViewNodes =
-            if (options.sessionReplay.isCaptureSurfaceViews) {
-              mutableListOf<ViewHierarchyNode.SurfaceViewHierarchyNode>()
-            } else {
-              null
-            }
-          root.traverse(viewHierarchy, options.sessionReplay, options.logger, surfaceViewNodes)
-
-          if (surfaceViewNodes.isNullOrEmpty()) {
-            executor.submit(
-              ReplayRunnable("screenshot_recorder.mask") {
-                applyMaskingAndNotify(
-                  root,
-                  viewHierarchy,
-                  resetUnstableCaptures = !changedDuringCapture,
+          // Release the frame gate if anything below throws before we hand work off to the
+          // executor — otherwise a single failure wedges captures forever.
+          try {
+            // TODO: disableAllMasking here and dont traverse?
+            val viewHierarchy = ViewHierarchyNode.fromView(root, null, 0, options.sessionReplay)
+            val surfaceViewNodes =
+              if (options.sessionReplay.isCaptureSurfaceViews) {
+                mutableListOf<ViewHierarchyNode.SurfaceViewHierarchyNode>()
+              } else {
+                null
+              }
+            root.traverse(viewHierarchy, options.sessionReplay, options.logger, surfaceViewNodes)
+
+            if (surfaceViewNodes.isNullOrEmpty()) {
+              val submitted =
+                executor.submit(
+                  ReplayRunnable("screenshot_recorder.mask") {
+                    try {
+                      applyMaskingAndNotify(
+                        root,
+                        viewHierarchy,
+                        resetUnstableCaptures = !changedDuringCapture,
+                      )
+                    } finally {
+                      finishFrame()
+                    }
+                  }
                 )
+              if (submitted == null) {
+                finishFrame()
               }
-            )
-          } else {
-            // Re-arm the recorder's contentChanged gate; SurfaceView redraws don't trigger
-            // ViewTreeObserver.OnDrawListener, so we'd otherwise emit the same frame forever.
-            markContentChanged()
-            captureSurfaceViews(
-              root,
-              surfaceViewNodes,
-              viewHierarchy,
-              resetUnstableCaptures = !changedDuringCapture,
-            )
+            } else {
+              // Re-arm the recorder's contentChanged gate; SurfaceView redraws don't trigger
+              // ViewTreeObserver.OnDrawListener, so we'd otherwise emit the same frame forever.
+              markContentChanged()
+              captureSurfaceViews(
+                root,
+                surfaceViewNodes,
+                viewHierarchy,
+                resetUnstableCaptures = !changedDuringCapture,
+              )
+            }
+          } catch (e: RuntimeException) {
+            // OEM View subclasses have been observed throwing during hierarchy traversal
+            // (e.g. Redmi's TextView NPE). Release the frame gate so a single bad frame
+            // doesn't wedge the recorder. Errors (OOM, LinkageError) intentionally propagate.
+            options.logger.log(WARNING, "Failed to process replay frame", e)
+            finishFrame()
           }
         },
         mainLooperHandler.handler,
@@ -143,6 +172,7 @@ internal class PixelCopyStrategy(
       options.logger.log(WARNING, "Failed to capture replay recording", e)
       unstableCaptures.set(0)
       lastCaptureSuccessful.set(false)
+      finishFrame()
     }
   }
 
@@ -272,37 +302,46 @@ internal class PixelCopyStrategy(
     windowY: Int,
     resetUnstableCaptures: Boolean,
   ) {
-    executor.submit(
-      ReplayRunnable("screenshot_recorder.composite") {
-        if (isClosed.get() || screenshot.isRecycled) {
-          options.logger.log(DEBUG, "PixelCopyStrategy is closed, skipping compositing")
-          recycleCaptures(captures)
-          return@ReplayRunnable
-        }
+    val submitted =
+      executor.submit(
+        ReplayRunnable("screenshot_recorder.composite") {
+          try {
+            if (isClosed.get() || screenshot.isRecycled) {
+              options.logger.log(DEBUG, "PixelCopyStrategy is closed, skipping compositing")
+              recycleCaptures(captures)
+              return@ReplayRunnable
+            }
 
-        for (capture in captures) {
-          if (capture == null) continue
-          if (capture.bitmap.isRecycled) continue
-
-          compositeSurfaceViewInto(
-            screenshotCanvas,
-            dstOverPaint,
-            tmpSrcRect,
-            tmpDstRect,
-            capture.bitmap,
-            capture.x,
-            capture.y,
-            windowX,
-            windowY,
-            config.scaleFactorX,
-            config.scaleFactorY,
-          )
-          capture.bitmap.recycle()
-        }
+            for (capture in captures) {
+              if (capture == null) continue
+              if (capture.bitmap.isRecycled) continue
+
+              compositeSurfaceViewInto(
+                screenshotCanvas,
+                dstOverPaint,
+                tmpSrcRect,
+                tmpDstRect,
+                capture.bitmap,
+                capture.x,
+                capture.y,
+                windowX,
+                windowY,
+                config.scaleFactorX,
+                config.scaleFactorY,
+              )
+              capture.bitmap.recycle()
+            }
 
-        applyMaskingAndNotify(root, viewHierarchy, resetUnstableCaptures)
-      }
-    )
+            applyMaskingAndNotify(root, viewHierarchy, resetUnstableCaptures)
+          } finally {
+            finishFrame()
+          }
+        }
+      )
+    if (submitted == null) {
+      recycleCaptures(captures)
+      finishFrame()
+    }
   }
 
   private fun recycleCaptures(captures: Array<SurfaceViewCapture?>) {
@@ -322,15 +361,57 @@ internal class PixelCopyStrategy(
   }
 
   override fun emitLastScreenshot() {
-    if (lastCaptureSuccessful() && !screenshot.isRecycled) {
-      screenshotRecorderCallback?.onScreenshotRecorded(screenshot)
+    if (!frameInFlight.compareAndSet(false, true)) {
+      return
+    }
+    if (!lastCaptureSuccessful() || screenshot.isRecycled) {
+      finishFrame()
+      return
+    }
+    // Submit to the executor so the downstream consumer's bitmap read (JPEG compress) runs inline
+    // on the worker thread while the gate is held, same as the masked capture path.
+    val submitted =
+      executor.submit(
+        ReplayRunnable("PixelCopyStrategy.emit") {
+          try {
+            screenshotRecorderCallback?.onScreenshotRecorded(screenshot)
+          } finally {
+            finishFrame()
+          }
+        }
+      )
+    if (submitted == null) {
+      finishFrame()
     }
   }
 
   override fun close() {
     isClosed.set(true)
     unstableCaptures.set(0)
-    executor.submit(
+    cleanUpIfIdle()
+  }
+
+  private fun finishFrame() {
+    frameInFlight.set(false)
+    if (isClosed.get()) {
+      cleanUpIfIdle()
+    }
+  }
+
+  /**
+   * Schedules cleanup only for the caller that owns the gate. Whoever wins [frameInFlight]'s CAS
+   * (close when no frame is running, or the finishFrame of the last in-flight frame after close)
+   * runs cleanup exactly once; a racing capture that took the gate loses the CAS and backs off, so
+   * we never recycle the shared screenshot while that capture is still using it.
+   */
+  private fun cleanUpIfIdle() {
+    if (frameInFlight.compareAndSet(false, true)) {
+      scheduleCleanup()
+    }
+  }
+
+  private fun scheduleCleanup() {
+    val cleanup =
       ReplayRunnable(
         "PixelCopyStrategy.close",
         {
@@ -344,7 +425,12 @@ internal class PixelCopyStrategy(
           maskRenderer.close()
         },
       )
-    )
+    // ReplayExecutorService.submit returns null only on genuine rejection (post-shutdown);
+    // inline execution on the worker thread returns a completed future. Fall back to running
+    // cleanup here so the bitmap + mask renderer are freed even when the executor is dead.
+    if (executor.submit(cleanup) == null) {
+      cleanup.run()
+    }
   }
 }
 

sentry-android-replay/src/main/java/io/sentry/android/replay/util/ReplayExecutorService.kt

@@ -4,6 +4,7 @@ import io.sentry.SentryLevel.ERROR
 import io.sentry.SentryOptions
 import java.util.concurrent.Future
 import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeUnit.MILLISECONDS
 
 /**
@@ -14,11 +15,20 @@ internal class ReplayExecutorService(
   private val delegate: ScheduledExecutorService,
   private val options: SentryOptions,
 ) : ScheduledExecutorService by delegate {
+  /**
+   * Submits [task] for execution and returns a [Future] describing what happened. The return value
+   * has three distinct outcomes callers can rely on:
+   * - [CompletedFuture] — the caller is already on the replay worker thread, so the task was run
+   *   inline before this method returned. Skips the queue.
+   * - A regular [Future] from the underlying [ScheduledExecutorService] — the task was queued and
+   *   will run asynchronously.
+   * - `null` — the underlying executor rejected the submission (typically because it has been shut
+   *   down). The task did NOT run; callers that need cleanup must handle it themselves.
+   */
   override fun submit(task: Runnable): Future<*>? {
     if (Thread.currentThread().name.startsWith("SentryReplayIntegration")) {
-      // we're already on the worker thread, no need to submit
       task.run()
-      return null
+      return CompletedFuture
     }
     return try {
       delegate.submit {
@@ -68,3 +78,16 @@ internal class ReplayExecutorService(
 }
 
 internal class ReplayRunnable(val taskName: String, delegate: Runnable) : Runnable by delegate
+
+/** A Future that represents an already-completed inline execution — never used as null. */
+internal object CompletedFuture : Future<Unit> {
+  override fun cancel(mayInterruptIfRunning: Boolean): Boolean = false
+
+  override fun isCancelled(): Boolean = false
+
+  override fun isDone(): Boolean = true
+
+  override fun get() {}
+
+  override fun get(timeout: Long, unit: TimeUnit) {}
+}