fix(core): Prevent lost wakeups in batch processors

CHANGELOG.md

@@ -30,6 +30,7 @@
 
 ### Fixes
 
+- Prevent logs and metrics from remaining queued after a flush scheduling race ([#5756](https://github.com/getsentry/sentry-java/pull/5756))
 - Fix main thread identification for tombstone (native crash) events ([#5742](https://github.com/getsentry/sentry-java/pull/5742))
 
 ### Dependencies

sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java

@@ -4,23 +4,21 @@
 import io.sentry.DataCategory;
 import io.sentry.ISentryClient;
 import io.sentry.ISentryExecutorService;
-import io.sentry.ISentryLifecycleToken;
 import io.sentry.SentryExecutorService;
 import io.sentry.SentryLevel;
 import io.sentry.SentryLogEvent;
 import io.sentry.SentryLogEvents;
 import io.sentry.SentryOptions;
 import io.sentry.clientreport.DiscardReason;
 import io.sentry.transport.ReusableCountLatch;
-import io.sentry.util.AutoClosableReentrantLock;
 import io.sentry.util.JsonSerializationUtils;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -37,9 +35,7 @@ public class LoggerBatchProcessor implements ILoggerBatchProcessor {
   private final @NotNull ISentryClient client;
   private final @NotNull Queue<SentryLogEvent> queue;
   private final @NotNull ISentryExecutorService executorService;
-  private volatile @Nullable Future<?> scheduledFlush;
-  private final @NotNull AutoClosableReentrantLock scheduleLock = new AutoClosableReentrantLock();
-  private volatile boolean hasScheduled = false;
+  private final @NotNull AtomicBoolean hasScheduled = new AtomicBoolean(false);
   private volatile boolean isShuttingDown = false;
 
   private final @NotNull ReusableCountLatch pendingCount = new ReusableCountLatch();
@@ -79,15 +75,15 @@ public void add(final @NotNull SentryLogEvent logEvent) {
     }
     pendingCount.increment();
     queue.offer(logEvent);
-    maybeSchedule(false, false);
+    maybeSchedule(false);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void close(final boolean isRestarting) {
     isShuttingDown = true;
     if (isRestarting) {
-      maybeSchedule(true, true);
+      maybeSchedule(true);
       executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
     } else {
       executorService.close(options.getShutdownTimeoutMillis());
@@ -97,33 +93,28 @@ public void close(final boolean isRestarting) {
     }
   }
 
-  private void maybeSchedule(boolean forceSchedule, boolean immediately) {
-    if (hasScheduled && !forceSchedule) {
+  @SuppressWarnings("FutureReturnValueIgnored")
+  private void maybeSchedule(boolean immediately) {
+    if (immediately) {
+      // any already scheduled task may be far in the future, we want to schedule something that
+      // runs right away
+      hasScheduled.set(true);
+    } else if (!hasScheduled.compareAndSet(false, true)) {
+      // was already true, no need to schedule again
       return;
     }
-    try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
-      final @Nullable Future<?> latestScheduledFlush = scheduledFlush;
-      if (forceSchedule
-          || latestScheduledFlush == null
-          || latestScheduledFlush.isDone()
-          || latestScheduledFlush.isCancelled()) {
-        hasScheduled = true;
-        final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
-        try {
-          scheduledFlush = executorService.schedule(new BatchRunnable(), flushAfterMs);
-        } catch (RejectedExecutionException e) {
-          hasScheduled = false;
-          options
-              .getLogger()
-              .log(SentryLevel.WARNING, "Logs batch processor flush task rejected", e);
-        }
-      }
+    final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
+    try {
+      executorService.schedule(new BatchRunnable(), flushAfterMs);
+    } catch (RejectedExecutionException e) {
+      hasScheduled.set(false);
+      options.getLogger().log(SentryLevel.WARNING, "Logs batch processor flush task rejected", e);
     }
   }
 
   @Override
   public void flush(long timeoutMillis) {
-    maybeSchedule(true, true);
+    maybeSchedule(true);
     try {
       pendingCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
@@ -134,12 +125,9 @@ public void flush(long timeoutMillis) {
 
   private void flush() {
     flushInternal();
-    try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
-      if (!queue.isEmpty()) {
-        maybeSchedule(true, false);
-      } else {
-        hasScheduled = false;
-      }
+    hasScheduled.set(false);
+    if (!queue.isEmpty()) {
+      maybeSchedule(false);
     }
   }
 

sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java

@@ -4,23 +4,21 @@
 import io.sentry.DataCategory;
 import io.sentry.ISentryClient;
 import io.sentry.ISentryExecutorService;
-import io.sentry.ISentryLifecycleToken;
 import io.sentry.SentryExecutorService;
 import io.sentry.SentryLevel;
 import io.sentry.SentryMetricsEvent;
 import io.sentry.SentryMetricsEvents;
 import io.sentry.SentryOptions;
 import io.sentry.clientreport.DiscardReason;
 import io.sentry.transport.ReusableCountLatch;
-import io.sentry.util.AutoClosableReentrantLock;
 import io.sentry.util.JsonSerializationUtils;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -35,9 +33,7 @@ public class MetricsBatchProcessor implements IMetricsBatchProcessor {
   private final @NotNull ISentryClient client;
   private final @NotNull Queue<SentryMetricsEvent> queue;
   private final @NotNull ISentryExecutorService executorService;
-  private volatile @Nullable Future<?> scheduledFlush;
-  private final @NotNull AutoClosableReentrantLock scheduleLock = new AutoClosableReentrantLock();
-  private volatile boolean hasScheduled = false;
+  private final @NotNull AtomicBoolean hasScheduled = new AtomicBoolean(false);
   private volatile boolean isShuttingDown = false;
 
   private final @NotNull ReusableCountLatch pendingCount = new ReusableCountLatch();
@@ -69,15 +65,15 @@ public void add(final @NotNull SentryMetricsEvent metricsEvent) {
     }
     pendingCount.increment();
     queue.offer(metricsEvent);
-    maybeSchedule(false, false);
+    maybeSchedule(false);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void close(final boolean isRestarting) {
     isShuttingDown = true;
     if (isRestarting) {
-      maybeSchedule(true, true);
+      maybeSchedule(true);
       executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
     } else {
       executorService.close(options.getShutdownTimeoutMillis());
@@ -87,33 +83,30 @@ public void close(final boolean isRestarting) {
     }
   }
 
-  private void maybeSchedule(boolean forceSchedule, boolean immediately) {
-    if (hasScheduled && !forceSchedule) {
+  @SuppressWarnings("FutureReturnValueIgnored")
+  private void maybeSchedule(boolean immediately) {
+    if (immediately) {
+      // any already scheduled task may be far in the future, we want to schedule something that
+      // runs right away
+      hasScheduled.set(true);
+    } else if (!hasScheduled.compareAndSet(false, true)) {
+      // was already true, no need to schedule again
       return;
     }
-    try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
-      final @Nullable Future<?> latestScheduledFlush = scheduledFlush;
-      if (forceSchedule
-          || latestScheduledFlush == null
-          || latestScheduledFlush.isDone()
-          || latestScheduledFlush.isCancelled()) {
-        hasScheduled = true;
-        final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
-        try {
-          scheduledFlush = executorService.schedule(new BatchRunnable(), flushAfterMs);
-        } catch (RejectedExecutionException e) {
-          hasScheduled = false;
-          options
-              .getLogger()
-              .log(SentryLevel.WARNING, "Metrics batch processor flush task rejected", e);
-        }
-      }
+    final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
+    try {
+      executorService.schedule(new BatchRunnable(), flushAfterMs);
+    } catch (RejectedExecutionException e) {
+      hasScheduled.set(false);
+      options
+          .getLogger()
+          .log(SentryLevel.WARNING, "Metrics batch processor flush task rejected", e);
     }
   }
 
   @Override
   public void flush(long timeoutMillis) {
-    maybeSchedule(true, true);
+    maybeSchedule(true);
     try {
       pendingCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
@@ -124,12 +117,9 @@ public void flush(long timeoutMillis) {
 
   private void flush() {
     flushInternal();
-    try (final @NotNull ISentryLifecycleToken ignored = scheduleLock.acquire()) {
-      if (!queue.isEmpty()) {
-        maybeSchedule(true, false);
-      } else {
-        hasScheduled = false;
-      }
+    hasScheduled.set(false);
+    if (!queue.isEmpty()) {
+      maybeSchedule(false);
     }
   }
 

sentry/src/test/java/io/sentry/logger/LoggerBatchProcessorTest.kt

@@ -1,5 +1,6 @@
 package io.sentry.logger
 
+import com.google.common.truth.Truth.assertThat
 import io.sentry.DataCategory
 import io.sentry.ISentryClient
 import io.sentry.SentryLogEvent
@@ -21,9 +22,30 @@ import kotlin.test.assertTrue
 import org.mockito.kotlin.argumentCaptor
 import org.mockito.kotlin.atLeast
 import org.mockito.kotlin.mock
+import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
 
 class LoggerBatchProcessorTest {
+  @Test
+  fun `schedules another flush after previous flush has run`() {
+    val mockClient = mock<ISentryClient>()
+    val mockExecutor = DeferredExecutorService()
+    val processor = LoggerBatchProcessor(SentryOptions(), mockClient, mockExecutor)
+
+    processor.add(SentryLogEvent(SentryId(), SentryNanotimeDate(), "first", SentryLogLevel.INFO))
+    mockExecutor.runAll()
+
+    processor.add(SentryLogEvent(SentryId(), SentryNanotimeDate(), "second", SentryLogLevel.INFO))
+    assertThat(mockExecutor.hasScheduledRunnables()).isTrue()
+    mockExecutor.runAll()
+
+    val captor = argumentCaptor<SentryLogEvents>()
+    verify(mockClient, times(2)).captureBatchedLogEvents(captor.capture())
+    assertThat(captor.allValues.flatMap { it.items }.map { it.body })
+      .containsExactly("first", "second")
+      .inOrder()
+  }
+
   @Test
   fun `drops log events after reaching MAX_QUEUE_SIZE limit`() {
     // given

sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt

@@ -1,5 +1,6 @@
 package io.sentry.metrics
 
+import com.google.common.truth.Truth.assertThat
 import io.sentry.DataCategory
 import io.sentry.ISentryClient
 import io.sentry.SentryMetricsEvent
@@ -20,9 +21,31 @@ import kotlin.test.assertTrue
 import org.mockito.kotlin.argumentCaptor
 import org.mockito.kotlin.atLeast
 import org.mockito.kotlin.mock
+import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
 
 class MetricsBatchProcessorTest {
+  @Test
+  fun `schedules another flush after previous flush has run`() {
+    val mockClient = mock<ISentryClient>()
+    val mockExecutor = DeferredExecutorService()
+    val processor = MetricsBatchProcessor(SentryOptions(), mockClient)
+    processor.injectForField("executorService", mockExecutor)
+
+    processor.add(SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "first", "gauge", 1.0))
+    mockExecutor.runAll()
+
+    processor.add(SentryMetricsEvent(SentryId(), SentryNanotimeDate(), "second", "gauge", 2.0))
+    assertThat(mockExecutor.hasScheduledRunnables()).isTrue()
+    mockExecutor.runAll()
+
+    val captor = argumentCaptor<SentryMetricsEvents>()
+    verify(mockClient, times(2)).captureBatchedMetricsEvents(captor.capture())
+    assertThat(captor.allValues.flatMap { it.items }.map { it.name })
+      .containsExactly("first", "second")
+      .inOrder()
+  }
+
   @Test
   fun `drops metrics events after reaching MAX_QUEUE_SIZE limit`() {
     // given