fix: coalesce pending session replay captures per window

.changeset/cyan-geese-tell.md

@@ -0,0 +1,5 @@
+---
+"posthog-android": patch
+---
+
+Skip redundant session replay captures while a window has a queued capture or unfinished PixelCopy callback.

posthog-android/src/main/java/com/posthog/android/replay/PostHogReplayIntegration.kt

@@ -334,8 +334,11 @@ public class PostHogReplayIntegration(
                                             return@onNextDraw
                                         }
 
-                                        executor.submit {
+                                        submitCapture(drawState) {
                                             try {
+                                                if (decorViews[decorView]?.drawState !== drawState) {
+                                                    return@submitCapture
+                                                }
                                                 generateSnapshot(WeakReference(decorView), WeakReference(window))
                                             } catch (e: Throwable) {
                                                 config.logger.log("Session Replay generateSnapshot failed: $e.")
@@ -373,6 +376,29 @@ public class PostHogReplayIntegration(
         }
     }
 
+    // Acquire before submitting, not on the worker: a busy worker must not accumulate
+    // redundant captures. Draw-time mask verification still runs for every draw.
+    internal fun submitCapture(
+        drawState: WindowDrawState,
+        capture: () -> Unit,
+    ): Boolean {
+        if (!drawState.tryScheduleCapture()) return false
+        try {
+            executor.submit {
+                try {
+                    capture()
+                } finally {
+                    drawState.finishScheduledCapture()
+                }
+            }
+        } catch (e: Throwable) {
+            drawState.finishScheduledCapture()
+            config.logger.log("Session Replay capture submission failed: $e.")
+            return false
+        }
+        return true
+    }
+
     private val onRootViewsChangedListener =
         OnRootViewsChangedListener { view, added ->
             addView(view, added)
@@ -612,7 +638,8 @@ public class PostHogReplayIntegration(
      *
      * The return value only means the capture was scheduled; [onResult] fires on
      * the capture thread with whether a frame was actually delivered — callers
-     * must treat that, not the return value, as the retry signal.
+     * must treat that, not the return value, as the retry signal. Returns false without
+     * calling [onResult] if this window already has a pending/in-flight capture.
      */
     @PostHogInternalReplayApi
     public fun captureSessionReplaySnapshot(
@@ -642,13 +669,14 @@ public class PostHogReplayIntegration(
                 return false
             }
             val window = decorView.phoneWindow ?: return false
-            if (decorViews[decorView] == null) {
+            val drawState = decorViews[decorView]?.drawState
+            if (drawState == null) {
                 // Not tracked yet (onDecorViewReady pending): generateSnapshot
                 // would bail silently — report failure so the caller retries
                 // and the first-of-episode reset is not consumed.
                 return false
             }
-            executor.submit {
+            return submitCapture(drawState) {
                 // A throwing onResult would land in the catch below and fire a
                 // second time — report exactly once per scheduled capture.
                 var resultReported = false
@@ -663,12 +691,12 @@ public class PostHogReplayIntegration(
                     // the capture thread: the reset mutates snapshot status
                     // fields that are otherwise only touched here, and a
                     // stale queued capture must not emit after the episode.
-                    if (!isStillValid()) {
+                    if (!isStillValid() || decorViews[decorView]?.drawState !== drawState) {
                         // The contract promises onResult for every scheduled
                         // capture; a silent self-drop would leave the caller's
                         // in-flight tracking latched forever.
                         report(false)
-                        return@submit
+                        return@submitCapture
                     }
                     if (forceFullSnapshot) {
                         decorViews[decorView]?.let { status ->
@@ -687,7 +715,6 @@ public class PostHogReplayIntegration(
                     report(false)
                 }
             }
-            return true
         } catch (e: Throwable) {
             config.logger.log("Session Replay bridge capture failed: $e.")
             return false
@@ -1643,6 +1670,7 @@ public class PostHogReplayIntegration(
         // We use the latch itself as the synchronization mechanism (await happens-before countDown)
         var callbackCompleted = false
 
+        drawState.beginPixelCopy()
         try {
             PixelCopy.request(window, bitmap, { copyResult ->
                 try {
@@ -1662,13 +1690,15 @@ public class PostHogReplayIntegration(
                     success = false
                 } finally {
                     callbackCompleted = true
+                    drawState.finishPixelCopy()
                     latch.countDown()
                 }
             }, handler)
         } catch (e: Throwable) {
             config.logger.log("Session Replay PixelCopy failed: $e.")
             success = false
             callbackCompleted = true
+            drawState.finishPixelCopy()
             latch.countDown()
         }
 

posthog-android/src/main/java/com/posthog/android/replay/internal/ViewTreeSnapshotStatus.kt

@@ -127,6 +127,33 @@ internal class WindowDrawState {
     private var nextCaptureId: Long = 0
     private var activeCapture: ActiveMaskCapture? = null
 
+    // Independent of mask/snapshot resets: stopping or rotating a session must not allow
+    // another task while the old worker or a timed-out PixelCopy callback is still running.
+    private var captureScheduled = false
+    private var pixelCopyInFlight = false
+
+    fun tryScheduleCapture(): Boolean =
+        synchronized(captureLock) {
+            if (captureScheduled || pixelCopyInFlight) {
+                false
+            } else {
+                captureScheduled = true
+                true
+            }
+        }
+
+    fun finishScheduledCapture() {
+        synchronized(captureLock) { captureScheduled = false }
+    }
+
+    fun beginPixelCopy() {
+        synchronized(captureLock) { pixelCopyInFlight = true }
+    }
+
+    fun finishPixelCopy() {
+        synchronized(captureLock) { pixelCopyInFlight = false }
+    }
+
     fun reset() {
         isOnDrawnCalled = false
         isOnlyAnimationRedraw = false

posthog-android/src/test/java/com/posthog/android/replay/PostHogReplayIntegrationTest.kt

@@ -40,10 +40,14 @@ import com.posthog.internal.PostHogQueue
 import com.posthog.internal.PostHogQueueInterface
 import com.posthog.internal.PostHogRemoteConfig
 import com.posthog.internal.PostHogSessionManager
+import curtains.Curtains
 import curtains.DispatchState
+import curtains.OnRootViewsChangedListener
 import org.junit.Rule
 import org.junit.rules.TemporaryFolder
 import org.junit.runner.RunWith
+import org.mockito.MockedStatic
+import org.mockito.Mockito.mockStatic
 import org.mockito.kotlin.any
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
@@ -65,6 +69,8 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
 import java.util.concurrent.Future
+import java.util.concurrent.FutureTask
+import java.util.concurrent.RejectedExecutionException
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
@@ -228,6 +234,144 @@ internal class PostHogReplayIntegrationTest {
         return PostHogReplayIntegration(context, config, MainHandler(), executor)
     }
 
+    // Robolectric resets WindowManagerGlobal between tests, but Curtains caches its old
+    // root list. Supply the actual activity explicitly rather than relying on that cache.
+    private fun mockCurtainsRoot(view: View): MockedStatic<Curtains> {
+        val mocked = mockStatic(Curtains::class.java)
+        mocked.`when`<List<View>> { Curtains.rootViews }.thenReturn(listOf(view))
+        mocked.`when`<MutableList<OnRootViewsChangedListener>> { Curtains.onRootViewsChangedListeners }
+            .thenReturn(mutableListOf())
+        return mocked
+    }
+
+    private class QueuedReplayExecutor(delegate: ExecutorService) : ExecutorService by delegate {
+        val tasks = mutableListOf<FutureTask<*>>()
+        var reject = false
+
+        override fun submit(task: Runnable): Future<*> {
+            if (reject) throw RejectedExecutionException("Test rejection")
+            return FutureTask(task, null).also { tasks.add(it) }
+        }
+    }
+
+    @Test
+    fun `draw requests are coalesced per window while capture is queued`() {
+        val executor = QueuedReplayExecutor(createReplayExecutor())
+        val tasks = executor.tasks
+        val config = configWithSampling(flagActive = true, samplingPasses = true)
+        config.sessionReplayConfig.throttleDelayMs = 0
+        val sut = getSutWithExecutor(config, executor)
+        val controller = Robolectric.buildActivity(Activity::class.java).setup()
+        val curtains = mockCurtainsRoot(controller.get().window.decorView)
+        sut.install(createPostHogFake())
+        try {
+            sut.start(resumeCurrent = true)
+            shadowOf(Looper.getMainLooper()).idle()
+            val status = assertNotNull(sut.decorViews[controller.get().window.decorView])
+            tasks.forEach { it.run() }
+            tasks.clear()
+
+            repeat(100) { status.listener.onDraw() }
+
+            assertEquals(1, tasks.size)
+            // Coalescing must not suppress the unthrottled draw/mask callback.
+            assertTrue(status.drawState.isOnDrawnCalled)
+
+            // An early bail (stopped before execution) must release the gate too.
+            sut.stop()
+            tasks.removeAt(0).run()
+            sut.start(resumeCurrent = true)
+            status.listener.onDraw()
+            assertEquals(1, tasks.size)
+        } finally {
+            sut.uninstall()
+            curtains.close()
+            controller.pause().stop().destroy()
+        }
+    }
+
+    @Test
+    fun `running capture blocks duplicates but not other windows`() {
+        val executor = createReplayExecutor()
+        val sut = getSutWithExecutor(PostHogAndroidConfig(API_KEY), executor)
+        val firstWindow = WindowDrawState()
+        val secondWindow = WindowDrawState()
+        val entered = CountDownLatch(1)
+        val release = CountDownLatch(1)
+        try {
+            assertTrue(
+                sut.submitCapture(firstWindow) {
+                    entered.countDown()
+                    assertTrue(release.await(2, TimeUnit.SECONDS))
+                },
+            )
+            assertTrue(entered.await(2, TimeUnit.SECONDS))
+            repeat(100) { assertFalse(sut.submitCapture(firstWindow) { error("Duplicate capture") }) }
+            assertTrue(sut.submitCapture(secondWindow) {})
+            assertFalse(sut.submitCapture(secondWindow) {})
+        } finally {
+            release.countDown()
+            awaitReplayExecutors()
+        }
+        assertTrue(sut.submitCapture(firstWindow) {})
+        assertTrue(sut.submitCapture(secondWindow) {})
+        awaitReplayExecutors()
+    }
+
+    @Test
+    fun `capture gate releases after rejection and task failure`() {
+        val executor = QueuedReplayExecutor(createReplayExecutor())
+        val sut = getSutWithExecutor(PostHogAndroidConfig(API_KEY), executor)
+        val drawState = WindowDrawState()
+        executor.reject = true
+        assertFalse(sut.submitCapture(drawState) {})
+        executor.reject = false
+        assertTrue(sut.submitCapture(drawState) { error("Capture failed") })
+        executor.tasks.removeAt(0).run()
+        assertTrue(sut.submitCapture(drawState) {})
+        executor.tasks.removeAt(0).run()
+    }
+
+    @Test
+    @OptIn(PostHogInternalReplayApi::class)
+    fun `bridge capture shares draw gate and reports once for accepted requests only`() {
+        val executor = QueuedReplayExecutor(createReplayExecutor())
+        val config = configWithSampling(flagActive = true, samplingPasses = true)
+        config.sessionReplayConfig.throttleDelayMs = 0
+        val sut = getSutWithExecutor(config, executor)
+        val controller = Robolectric.buildActivity(Activity::class.java).setup()
+        val curtains = mockCurtainsRoot(controller.get().window.decorView)
+        sut.install(createPostHogFake())
+        try {
+            sut.start(resumeCurrent = true)
+            shadowOf(Looper.getMainLooper()).idle()
+            val status = assertNotNull(sut.decorViews[controller.get().window.decorView])
+            executor.tasks.forEach { it.run() }
+            executor.tasks.clear()
+            val results = mutableListOf<Boolean>()
+
+            assertTrue(sut.captureSessionReplaySnapshot(null, true, { false }) { results.add(it) })
+            assertFalse(sut.captureSessionReplaySnapshot(null, true, { true }) { error("Not scheduled") })
+            status.listener.onDraw()
+            assertEquals(1, executor.tasks.size)
+            assertTrue(results.isEmpty())
+            executor.tasks.removeAt(0).run()
+            assertEquals(listOf(false), results)
+
+            status.listener.onDraw()
+            assertFalse(sut.captureSessionReplaySnapshot(null, true, { true }) { error("Not scheduled") })
+            // Uninstalling with work queued must still let that work self-drop and release.
+            sut.uninstall()
+            executor.tasks.removeAt(0).run()
+            assertTrue(status.drawState.tryScheduleCapture())
+            status.drawState.finishScheduledCapture()
+        } finally {
+            sut.uninstall()
+            curtains.close()
+            controller.pause().stop().destroy()
+        }
+    }
+
     // currentTimeMillis() on Android does a network-time lookup; count how often the touch path
     // invokes it so we can prove it is skipped when replay is inactive.
     private class CountingDateProvider(val calls: AtomicInteger) : PostHogDateProvider {
@@ -1684,8 +1828,63 @@ internal class PostHogReplayIntegrationTest {
             fx.sut.generateSnapshot(WeakReference(decorView), WeakReference(mock<Window>()))
 
             assertEquals(0, fake.captures)
+            val drawState = fx.sut.decorViews[decorView]!!.drawState
+            assertTrue(drawState.tryScheduleCapture())
+            drawState.finishScheduledCapture()
+        } finally {
+            fx.sut.uninstall()
+        }
+    }
+
+    @Implements(PixelCopy::class)
+    class DelayedShadowPixelCopy {
+        companion object {
+            var callback: PixelCopy.OnPixelCopyFinishedListener? = null
+
+            @JvmStatic
+            @Implementation
+            fun request(
+                window: Window,
+                bitmap: Bitmap,
+                listener: PixelCopy.OnPixelCopyFinishedListener,
+                handler: Handler,
+            ) {
+                callback = listener
+            }
+        }
+    }
+
+    @Test
+    @Config(sdk = [26], shadows = [DelayedShadowPixelCopy::class])
+    fun `timed out PixelCopy keeps capture gate closed until callback completes`() {
+        val (fx, fake) = screenshotFixture()
+        val controller = Robolectric.buildActivity(Activity::class.java).setup()
+        try {
+            shadowOf(Looper.getMainLooper()).idle()
+            val window = controller.get().window
+            val decorView = window.decorView
+            makeWindowVisible(decorView)
+            val status = ViewTreeSnapshotStatus(mock<NextDrawListener>())
+            fx.sut.decorViews[decorView] = status
+            assertTrue(status.drawState.tryScheduleCapture())
+
+            assertFalse(fx.sut.generateSnapshot(WeakReference(decorView), WeakReference(window)))
+            status.drawState.finishScheduledCapture()
+            assertNotNull(DelayedShadowPixelCopy.callback)
+            assertFalse(status.drawState.tryScheduleCapture())
+            assertEquals(0, fake.captures)
+
+            // Stop/reset cannot release the gate while Android still owns the bitmap.
+            fx.sut.stop()
+            assertFalse(status.drawState.tryScheduleCapture())
+            DelayedShadowPixelCopy.callback!!.onPixelCopyFinished(PixelCopy.SUCCESS)
+            assertTrue(status.drawState.tryScheduleCapture())
+            status.drawState.finishScheduledCapture()
+            assertEquals(0, fake.captures)
         } finally {
+            DelayedShadowPixelCopy.callback = null
             fx.sut.uninstall()
+            controller.pause().stop().destroy()
         }
     }