perf(core): Coalesce scope writes on submit instead of a timer (JAVA-628)

Scope-persistence flushes were debounced 100ms behind a scheduled task.
The debounce was unnecessary: the Sentry executor is single-threaded, so a
submitted flush task already sits in the queue long enough for mutations
arriving behind it to be folded into the same write. That is exactly the
window that matters, since the queue is deepest during startup.

Submit the flush instead of scheduling it. Coalescing now tracks executor
load rather than a fixed delay, which closes the data-loss window the
debounce introduced. Guard against the executor rejecting the task without
throwing once its queue is full, which would otherwise leave the pending
flag set and stop scope persistence for the rest of the process.

Also record why resetCache() deliberately leaves pending mutations alone:
they only ever hold values from the current process, so dropping them
would lose scope state set during init rather than clearing the previous
run's data.

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

Author: runningcode

CHANGELOG.md

@@ -14,7 +14,7 @@
 ### Performance
 
 - Batch and coalesce scope-persistence disk writes to reduce startup cost ([#5791](https://github.com/getsentry/sentry-java/pull/5791))
-  - Scope mutations are now coalesced (latest value per field) and breadcrumbs are appended in batches behind a single fsync, instead of one synchronous disk write per mutation. Persisted scope data (used to enrich crash/ANR events on the next launch) may now lag real-time changes by up to ~100 ms.
+  - Scope mutations are now coalesced (latest value per field) and breadcrumbs are appended in batches behind a single fsync, instead of one synchronous disk write per mutation.
 - Reduce the number of SDK threads: `LifecycleWatcher` now schedules the session-end task on the shared timer executor instead of creating a dedicated `java.util.Timer` thread ([#5819](https://github.com/getsentry/sentry-java/pull/5819))
 
 ## 8.50.1

sentry/src/main/java/io/sentry/cache/PersistingScopeObserver.java

@@ -34,6 +34,7 @@
 import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -43,15 +44,6 @@ public final class PersistingScopeObserver extends ScopeObserverAdapter {
 
   private static final Charset UTF_8 = Charset.forName("UTF-8");
 
-  /**
-   * How long scope mutations are coalesced before being flushed to disk. Rather than writing on
-   * every mutation, we keep only the latest value per file (and buffer breadcrumbs) and flush them
-   * together after this delay. This trades a small data-loss window (mutations from the last
-   * ~{@value #FLUSH_AFTER_MS} ms before the process dies) for far fewer disk writes and fsyncs,
-   * which is significant during startup.
-   */
-  static final long FLUSH_AFTER_MS = 100;
-
   /** Sentinel value marking a file that should be deleted rather than written on the next flush. */
   private static final Object DELETE_MARKER = new Object();
 
@@ -136,7 +128,7 @@ public void toStream(Breadcrumb value, OutputStream sink) throws IOException {
   // Breadcrumbs buffered since the last flush, appended together behind a single fsync.
   private final @NotNull Queue<Breadcrumb> pendingBreadcrumbs = new ConcurrentLinkedQueue<>();
   private final @NotNull AtomicBoolean pendingBreadcrumbsClear = new AtomicBoolean(false);
-  private final @NotNull AtomicBoolean hasScheduledFlush = new AtomicBoolean(false);
+  private final @NotNull AtomicBoolean hasPendingFlush = new AtomicBoolean(false);
 
   public PersistingScopeObserver(final @NotNull SentryOptions options) {
     this.options = options;
@@ -153,7 +145,7 @@ public void addBreadcrumb(@NotNull Breadcrumb crumb) {
       return;
     }
     pendingBreadcrumbs.offer(crumb);
-    scheduleFlush();
+    requestFlush();
   }
 
   @Override
@@ -167,7 +159,7 @@ public void setBreadcrumbs(@NotNull Collection<Breadcrumb> breadcrumbs) {
       // drop breadcrumbs buffered before the clear; anything added after it is enqueued again
       pendingBreadcrumbs.clear();
       pendingBreadcrumbsClear.set(true);
-      scheduleFlush();
+      requestFlush();
     }
   }
 
@@ -226,31 +218,42 @@ private void enqueue(final @NotNull String fileName, final @Nullable Object enti
     }
     // latest value wins; a null entity means the file should be deleted on the next flush
     pendingWrites.put(fileName, entity == null ? DELETE_MARKER : entity);
-    scheduleFlush();
+    requestFlush();
   }
 
-  @SuppressWarnings("FutureReturnValueIgnored")
-  private void scheduleFlush() {
-    if (!hasScheduledFlush.compareAndSet(false, true)) {
-      // a flush is already scheduled; it will pick up the latest pending state
+  /**
+   * Queues a flush unless one is already queued. Coalescing comes from the flush task sitting in
+   * the executor queue: every mutation that arrives before it runs is folded into the same write.
+   * The executor is single-threaded, so during startup — when mutations are frequent and the queue
+   * is deep — that window covers many mutations.
+   */
+  private void requestFlush() {
+    if (!hasPendingFlush.compareAndSet(false, true)) {
+      // a flush is already queued; it will pick up the latest pending state
       return;
     }
     try {
-      options.getExecutorService().schedule(this::flushOnExecutor, FLUSH_AFTER_MS);
+      final @NotNull Future<?> future = options.getExecutorService().submit(this::flushOnExecutor);
+      if (future.isCancelled()) {
+        // the executor rejects tasks without throwing once its queue is full, so clear the flag or
+        // no later mutation would ever be able to queue a flush again
+        hasPendingFlush.set(false);
+      }
     } catch (Throwable e) {
-      hasScheduledFlush.set(false);
-      options.getLogger().log(ERROR, "Scope persistence flush could not be scheduled", e);
+      hasPendingFlush.set(false);
+      options.getLogger().log(ERROR, "Scope persistence flush could not be submitted", e);
     }
   }
 
   private void flushOnExecutor() {
     runSafely(this::flushPending);
-    hasScheduledFlush.set(false);
-    // reschedule if mutations arrived while we were flushing
+    // clear the flag before re-checking, otherwise a mutation landing between the drain and the
+    // clear would see a flush still queued and be left with nobody to write it
+    hasPendingFlush.set(false);
     if (!pendingWrites.isEmpty()
         || !pendingBreadcrumbs.isEmpty()
         || pendingBreadcrumbsClear.get()) {
-      scheduleFlush();
+      requestFlush();
     }
   }
 
@@ -353,6 +356,9 @@ public static <T> void store(
    * I/O and should be called from a background thread.
    */
   public void resetCache() {
+    // NOTE: pending mutations are deliberately left alone. They only ever hold values from the
+    // current process, which is exactly what this reset is clearing the way for; dropping them
+    // would lose scope state set during init.
     // since it keeps a reference to the file and we cannot delete it, breadcrumbs we just clear
     try {
       final @NotNull ObjectQueue<Breadcrumb> queue = breadcrumbsQueue.getValue();

sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt

@@ -30,7 +30,7 @@ class PersistingScopeObserverBatchingTest {
     read(options, BREADCRUMBS_FILENAME, List::class.java) as List<Breadcrumb>
 
   @Test
-  fun `defers writes until the scheduled flush runs`() {
+  fun `defers writes until the flush runs`() {
     val executor = DeferredExecutorService()
     val sut = getSut(executor)
 
@@ -80,6 +80,18 @@ class PersistingScopeObserverBatchingTest {
     assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("kept")
   }
 
+  @Test
+  fun `resetCache keeps pending mutations from the current process`() {
+    val executor = DeferredExecutorService()
+    val sut = getSut(executor)
+
+    sut.setTransaction("SetDuringInit")
+    sut.resetCache()
+    executor.runAll()
+
+    assertThat(sut.readTransaction()).isEqualTo("SetDuringInit")
+  }
+
   @Test
   fun `flush writes pending state synchronously`() {
     val executor = DeferredExecutorService()