fix(replay): Keep replays of this process in the startup cleanup

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Fixes
 
 - Session Replay: Fix a native crash (`SIGABRT` on the RenderThread) when stopping a replay that uses the canvas screenshot strategy ([#6342](https://github.com/getsentry/sentry-java/pull/6342))
+- Session Replay: Keep the segment videos of a replay that stopped before the startup cleanup of previous replays ran ([#6343](https://github.com/getsentry/sentry-java/pull/6343))
 - Report the number of dropped spans in client reports when a batched span envelope item is discarded, instead of a quantity of 1 (e.g. for Flutter span streaming) ([#6265](https://github.com/getsentry/sentry-java/pull/6265))
 - Session Replay: Keep the flushed video and do not reuse its segment id when `stop()` is called right after `flush()` in buffer mode ([#6340](https://github.com/getsentry/sentry-java/pull/6340))
 - Keep sending events from Android 7.1.0 and older devices after Sentry changes its TLS certificate authority ([#6227](https://github.com/getsentry/sentry-java/pull/6227))

sentry-android-replay/src/main/java/io/sentry/android/replay/ReplayIntegration.kt

@@ -59,6 +59,7 @@ import java.io.Closeable
 import java.io.File
 import java.util.Date
 import java.util.LinkedList
+import java.util.concurrent.CopyOnWriteArraySet
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
 import java.util.concurrent.ThreadFactory
@@ -85,6 +86,11 @@ public class ReplayIntegration(
       SentryIntegrationPackageStorage.getInstance()
         .addPackage("maven:io.sentry:sentry-android-replay", BuildConfig.VERSION_NAME)
     }
+
+    // Replays recorded by this process. Their folders hold segment videos the transport may still
+    // read after the replay stopped, so the startup cleanup of previous replays must keep them.
+    // Shared across instances, because a second SDK init registers a new ReplayIntegration.
+    val replayIdsOfThisProcess = CopyOnWriteArraySet<String>()
   }
 
   // needed for the Java's call site
@@ -247,6 +253,9 @@ public class ReplayIntegration(
     recorder?.start()
     strategy.start()
     val replayId: SentryId? = strategy.currentReplayId
+    if (replayId != null && replayId != SentryId.EMPTY_ID) {
+      replayIdsOfThisProcess += replayId.toString()
+    }
     state.set(
       ReplayState(
         generation = current.generation + 1,
@@ -638,15 +647,13 @@ public class ReplayIntegration(
   }
 
   private fun cleanupReplays(unfinishedReplayId: String = "") {
-    // clean up old replays
+    // clean up old replays, but keep the replays of this process and the previous process's
+    // unfinished replay (its folder is cleaned up after its envelope is assembled)
+    val keptReplayIds = replayIdsOfThisProcess + listOfNotNull(unfinishedReplayId.ifBlank { null })
     options.cacheDirPath?.let { cacheDir ->
       File(cacheDir).listFiles()?.forEach { file ->
         val name = file.name
-        if (
-          name.startsWith("replay_") &&
-            !name.contains(replayId.toString()) &&
-            !(unfinishedReplayId.isNotBlank() && name.contains(unfinishedReplayId))
-        ) {
+        if (name.startsWith("replay_") && keptReplayIds.none { name.contains(it) }) {
           FileUtils.deleteRecursively(file)
         }
       }
@@ -663,7 +670,12 @@ public class ReplayIntegration(
             return@submitSafely
           }
       val previousReplayId = SentryId(previousReplayIdString)
-      if (previousReplayId == SentryId.EMPTY_ID) {
+      // when this task runs late, the persisted replay id can already be one this process recorded,
+      // which is not a previous replay to recover
+      if (
+        previousReplayId == SentryId.EMPTY_ID ||
+          previousReplayId.toString() in replayIdsOfThisProcess
+      ) {
         cleanupReplays()
         return@submitSafely
       }

sentry-android-replay/src/test/java/io/sentry/android/replay/ReplayIntegrationTest.kt

@@ -1263,6 +1263,95 @@ class ReplayIntegrationTest {
     assertFalse(evenOlderReplay.exists())
   }
 
+  @Test
+  fun `delayed startup cleanup keeps replays recorded by this process`() {
+    val replayId = SentryId()
+    fixture.options.cacheDirPath = tmpDir.newFolder().absolutePath
+    val olderReplay =
+      File(fixture.options.cacheDirPath, "replay_${SentryId()}").also { it.mkdirs() }
+    // hold back the startup work, like a busy executor does
+    val deferredTasks = mutableListOf<Runnable>()
+    fixture.options.executorService = mock {
+      doAnswer { deferredTasks += it.arguments[0] as Runnable }
+        .whenever(mock)
+        .submit(any<Runnable>())
+    }
+
+    val captureStrategy = mock<CaptureStrategy> { on { currentReplayId }.thenReturn(replayId) }
+    val replay = fixture.getSut(context, replayCaptureStrategyProvider = { captureStrategy })
+    replay.register(fixture.scopes, fixture.options)
+    replay.startBuffering()
+    // a flushed segment video that is still waiting for the transport
+    val stoppedReplay =
+      File(fixture.options.cacheDirPath, "replay_$replayId").also {
+        it.mkdirs()
+        File(it, "0.mp4").writeBytes(byteArrayOf(1, 2, 3))
+      }
+    replay.stop()
+
+    deferredTasks.forEach { it.run() }
+
+    assertTrue(File(stoppedReplay, "0.mp4").exists())
+    assertFalse(olderReplay.exists())
+  }
+
+  @Test
+  fun `second SDK init keeps replays recorded by the first one`() {
+    val firstReplayId = SentryId()
+    fixture.options.cacheDirPath = tmpDir.newFolder().absolutePath
+    val firstStrategy = mock<CaptureStrategy> { on { currentReplayId }.thenReturn(firstReplayId) }
+    val first = fixture.getSut(context, replayCaptureStrategyProvider = { firstStrategy })
+    first.register(fixture.scopes, fixture.options)
+    first.startBuffering()
+    val firstReplay =
+      File(fixture.options.cacheDirPath, "replay_$firstReplayId").also {
+        it.mkdirs()
+        File(it, "0.mp4").writeBytes(byteArrayOf(1, 2, 3))
+      }
+    first.stop()
+    first.close()
+
+    // a new Sentry.init registers a new integration, which runs the startup cleanup again
+    val secondStrategy = mock<CaptureStrategy> { on { currentReplayId }.thenReturn(SentryId()) }
+    val second = fixture.getSut(context, replayCaptureStrategyProvider = { secondStrategy })
+    second.register(fixture.scopes, fixture.options)
+
+    assertTrue(File(firstReplay, "0.mp4").exists())
+  }
+
+  @Test
+  fun `delayed startup task does not finalize a replay recorded by this process`() {
+    val replayId = SentryId()
+    fixture.options.cacheDirPath = tmpDir.newFolder().absolutePath
+    fixture.options.addScopeObserver(PersistingScopeObserver(fixture.options))
+    val deferredTasks = mutableListOf<Runnable>()
+    fixture.options.executorService = mock {
+      doAnswer { deferredTasks += it.arguments[0] as Runnable }
+        .whenever(mock)
+        .submit(any<Runnable>())
+    }
+
+    val captureStrategy = mock<CaptureStrategy> { on { currentReplayId }.thenReturn(replayId) }
+    val replay = fixture.getSut(context, replayCaptureStrategyProvider = { captureStrategy })
+    replay.register(fixture.scopes, fixture.options)
+    replay.startBuffering()
+    val currentReplay =
+      File(fixture.options.cacheDirPath, "replay_$replayId").also {
+        it.mkdirs()
+        File(it, "0.mp4").writeBytes(byteArrayOf(1, 2, 3))
+      }
+    // the scope observer persisted this replay's id after a flush, before the startup task ran
+    File(fixture.options.cacheDirPath, PersistingScopeObserver.SCOPE_CACHE).also {
+      it.mkdirs()
+      File(it, PersistingScopeObserver.REPLAY_FILENAME).writeText("\"$replayId\"")
+    }
+
+    deferredTasks.forEach { it.run() }
+
+    assertTrue(File(currentReplay, "0.mp4").exists())
+    verify(fixture.scopes, never()).captureReplay(any(), any())
+  }
+
   @Test
   fun `onScreenshotRecorded supplies screen from scope to replay cache`() {
     val captureStrategy =