diff --git a/CHANGELOG.md b/CHANGELOG.md index a3baff8d6b2..a6600fda921 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java b/sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java index 81ae5b73c1a..71877c21dae 100644 --- a/sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java +++ b/sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java @@ -4,7 +4,6 @@ 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; @@ -12,15 +11,14 @@ 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 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,7 +75,7 @@ public void add(final @NotNull SentryLogEvent logEvent) { } pendingCount.increment(); queue.offer(logEvent); - maybeSchedule(false, false); + maybeSchedule(false); } @SuppressWarnings("FutureReturnValueIgnored") @@ -87,7 +83,7 @@ public void add(final @NotNull SentryLogEvent logEvent) { 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); } } diff --git a/sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java b/sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java index 2d5c78e5e89..3c744dbe3c5 100644 --- a/sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java +++ b/sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java @@ -4,7 +4,6 @@ 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; @@ -12,15 +11,14 @@ 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 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,7 +65,7 @@ public void add(final @NotNull SentryMetricsEvent metricsEvent) { } pendingCount.increment(); queue.offer(metricsEvent); - maybeSchedule(false, false); + maybeSchedule(false); } @SuppressWarnings("FutureReturnValueIgnored") @@ -77,7 +73,7 @@ public void add(final @NotNull SentryMetricsEvent metricsEvent) { 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); } } diff --git a/sentry/src/test/java/io/sentry/logger/LoggerBatchProcessorTest.kt b/sentry/src/test/java/io/sentry/logger/LoggerBatchProcessorTest.kt index 73f2f03ebc3..01b8afdb4a3 100644 --- a/sentry/src/test/java/io/sentry/logger/LoggerBatchProcessorTest.kt +++ b/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() + 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() + 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 diff --git a/sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt b/sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt index 5ca1d0fe87a..d8320d9b1a6 100644 --- a/sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt +++ b/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() + 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() + 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