refactor(replay): move persistingExecutor ownership to ReplayIntegration

Move persistingExecutor out of BaseCaptureStrategy and into
ReplayIntegration, passing it as a constructor argument to CaptureStrategy
subclasses. Shut it down in ReplayIntegration.close() alongside
replayExecutor so executor lifecycle is managed in one place.

Author: tsushanth

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

@@ -111,6 +111,11 @@ public class ReplayIntegration(
     val delegate = Executors.newSingleThreadScheduledExecutor(ReplayExecutorServiceThreadFactory())
     ReplayExecutorService(delegate, options)
   }
+  private val persistingExecutor by lazy {
+    val delegate =
+      Executors.newSingleThreadScheduledExecutor(ReplayPersistingExecutorServiceThreadFactory())
+    ReplayExecutorService(delegate, options)
+  }
 
   internal val isEnabled = AtomicBoolean(false)
   internal val isManualPause = AtomicBoolean(false)
@@ -192,6 +197,7 @@ public class ReplayIntegration(
               scopes,
               dateProvider,
               replayExecutor,
+              persistingExecutor,
               replayCacheProvider,
             )
           } else {
@@ -201,6 +207,7 @@ public class ReplayIntegration(
               dateProvider,
               random,
               replayExecutor,
+              persistingExecutor,
               replayCacheProvider,
             )
           }
@@ -374,6 +381,7 @@ public class ReplayIntegration(
       recorder = null
       rootViewsSpy.close()
       replayExecutor.shutdown()
+      persistingExecutor.shutdown()
       lifecycle.currentState = CLOSED
     }
   }
@@ -554,4 +562,14 @@ public class ReplayIntegration(
       return ret
     }
   }
+
+  private class ReplayPersistingExecutorServiceThreadFactory : ThreadFactory {
+    private var cnt = 0
+
+    override fun newThread(r: Runnable): Thread {
+      val ret = Thread(r, "SentryReplayPersister-" + cnt++)
+      ret.setDaemon(true)
+      return ret
+    }
+  }
 }

sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt

@@ -25,7 +25,6 @@ import io.sentry.android.replay.ScreenshotRecorderConfig
 import io.sentry.android.replay.capture.CaptureStrategy.Companion.createSegment
 import io.sentry.android.replay.capture.CaptureStrategy.ReplaySegment
 import io.sentry.android.replay.gestures.ReplayGestureConverter
-import io.sentry.android.replay.util.ReplayExecutorService
 import io.sentry.android.replay.util.ReplayRunnable
 import io.sentry.protocol.SentryId
 import io.sentry.rrweb.RRWebEvent
@@ -34,9 +33,7 @@ import java.io.File
 import java.util.Date
 import java.util.Deque
 import java.util.concurrent.ConcurrentLinkedDeque
-import java.util.concurrent.Executors
 import java.util.concurrent.ScheduledExecutorService
-import java.util.concurrent.ThreadFactory
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
@@ -50,6 +47,7 @@ internal abstract class BaseCaptureStrategy(
   private val scopes: IScopes?,
   private val dateProvider: ICurrentDateProvider,
   protected val replayExecutor: ScheduledExecutorService,
+  protected val persistingExecutor: ScheduledExecutorService,
   private val replayCacheProvider: ((replayId: SentryId) -> ReplayCache)? = null,
 ) : CaptureStrategy {
   internal companion object {
@@ -58,21 +56,6 @@ internal abstract class BaseCaptureStrategy(
     private const val MAX_TRACE_IDS = 100
   }
 
-  // Explicit holder so we can detect whether the executor was ever initialised and shut it down
-  // without initialising it as a side-effect (which is what a plain `lazy` delegate would do).
-  private var persistingExecutorHolder: ScheduledExecutorService? = null
-  private val persistingExecutor: ScheduledExecutorService
-    get() {
-      return persistingExecutorHolder
-        ?: run {
-          val delegate =
-            Executors.newSingleThreadScheduledExecutor(
-              ReplayPersistingExecutorServiceThreadFactory()
-            )
-          ReplayExecutorService(delegate, options).also { persistingExecutorHolder = it }
-        }
-    }
-
   private val gestureConverter = ReplayGestureConverter(dateProvider)
 
   protected val isTerminating = AtomicBoolean(false)
@@ -133,13 +116,6 @@ internal abstract class BaseCaptureStrategy(
     replayStartTimestamp.set(0)
     segmentTimestamp = null
     currentReplayId = SentryId.EMPTY_ID
-    // Shut down the persisting executor only if it was actually initialised.  We must not call the
-    // blocking ReplayExecutorService.shutdown() here because stop() may run on the main thread,
-    // and blocking there risks an ANR.  shutdownNow() is non-blocking: it cancels queued tasks and
-    // interrupts any running one, which is acceptable because the tasks are best-effort persistence
-    // writes and the cache has already been closed above.
-    persistingExecutorHolder?.shutdownNow()
-    persistingExecutorHolder = null
   }
 
   protected fun createSegmentInternal(
@@ -209,16 +185,6 @@ internal abstract class BaseCaptureStrategy(
     }
   }
 
-  private class ReplayPersistingExecutorServiceThreadFactory : ThreadFactory {
-    private var cnt = 0
-
-    override fun newThread(r: Runnable): Thread {
-      val ret = Thread(r, "SentryReplayPersister-" + cnt++)
-      ret.setDaemon(true)
-      return ret
-    }
-  }
-
   private inline fun <T> persistableAtomicNullable(
     initialValue: T? = null,
     propertyName: String,

sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BufferCaptureStrategy.kt

@@ -33,13 +33,15 @@ internal class BufferCaptureStrategy(
   private val dateProvider: ICurrentDateProvider,
   private val random: Random,
   executor: ScheduledExecutorService,
+  persistingExecutor: ScheduledExecutorService,
   replayCacheProvider: ((replayId: SentryId) -> ReplayCache)? = null,
 ) :
   BaseCaptureStrategy(
     options,
     scopes,
     dateProvider,
     executor,
+    persistingExecutor,
     replayCacheProvider = replayCacheProvider,
   ) {
   // TODO: capture envelopes for buffered segments instead, but don't send them until buffer is
@@ -150,8 +152,10 @@ internal class BufferCaptureStrategy(
       )
       return this
     }
-    // we hand over replayExecutor to the new strategy to preserve order of execution
-    val captureStrategy = SessionCaptureStrategy(options, scopes, dateProvider, replayExecutor)
+    // we hand over replayExecutor and persistingExecutor to the new strategy to preserve order of
+    // execution
+    val captureStrategy =
+      SessionCaptureStrategy(options, scopes, dateProvider, replayExecutor, persistingExecutor)
     captureStrategy.recorderConfig = recorderConfig
     captureStrategy.start(
       segmentId = currentSegment,

sentry-android-replay/src/main/java/io/sentry/android/replay/capture/SessionCaptureStrategy.kt

@@ -21,8 +21,17 @@ internal class SessionCaptureStrategy(
   private val scopes: IScopes?,
   private val dateProvider: ICurrentDateProvider,
   executor: ScheduledExecutorService,
+  persistingExecutor: ScheduledExecutorService,
   replayCacheProvider: ((replayId: SentryId) -> ReplayCache)? = null,
-) : BaseCaptureStrategy(options, scopes, dateProvider, executor, replayCacheProvider) {
+) :
+  BaseCaptureStrategy(
+    options,
+    scopes,
+    dateProvider,
+    executor,
+    persistingExecutor,
+    replayCacheProvider,
+  ) {
   internal companion object {
     private const val TAG = "SessionCaptureStrategy"
   }

sentry-android-replay/src/test/java/io/sentry/android/replay/capture/BufferCaptureStrategyTest.kt

@@ -111,6 +111,12 @@ class BufferCaptureStrategyTest {
             null
           }
         },
+        mock {
+          whenever(it.submit(any<Runnable>())).doAnswer { invocation ->
+            (invocation.arguments[0] as Runnable).run()
+            null
+          }
+        },
       ) { _ ->
         replayCache
       }

sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt

@@ -123,6 +123,14 @@ class SessionCaptureStrategyTest {
             .whenever(it)
             .submit(any<Runnable>())
         },
+        mock {
+          doAnswer { invocation ->
+              (invocation.arguments[0] as Runnable).run()
+              null
+            }
+            .whenever(it)
+            .submit(any<Runnable>())
+        },
       ) { _ ->
         replayCache
       }