From 6bf6c7f4634096df2369965044c5f7ebd3bf8cd2 Mon Sep 17 00:00:00 2001 From: Alexander Dinauer Date: Wed, 28 Oct 2026 13:47:30 +0100 Subject: [PATCH] feat(core): Add scope-based duplicate event processor Correlate repeated throwable captures by exact mechanism source within a short bounded window. Keep state weak, scope-forkable, and atomic so reusable singleton exceptions are not suppressed across unrelated occurrences. Refs GH-6365 Co-Authored-By: Claude --- ...DuplicateEventDetectionEventProcessor.java | 137 ++++++++ ...plicateEventDetectionEventProcessorTest.kt | 321 ++++++++++++++++++ 2 files changed, 458 insertions(+) create mode 100644 sentry/src/main/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessor.java create mode 100644 sentry/src/test/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessorTest.kt diff --git a/sentry/src/main/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessor.java b/sentry/src/main/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessor.java new file mode 100644 index 0000000000..069f12695c --- /dev/null +++ b/sentry/src/main/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessor.java @@ -0,0 +1,137 @@ +package io.sentry; + +import io.sentry.exception.ExceptionMechanismException; +import io.sentry.internal.eventprocessor.SentryEventProcessor; +import io.sentry.time.MonotonicTicker; +import java.util.Collections; +import java.util.Iterator; +import java.util.Map; +import java.util.WeakHashMap; +import java.util.concurrent.TimeUnit; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +final class ScopeBasedDuplicateEventDetectionEventProcessor implements SentryEventProcessor { + static final long DEFAULT_CORRELATION_WINDOW_NANOS = TimeUnit.SECONDS.toNanos(1); + static final int DEFAULT_MAX_TRACKED_THROWABLES = 1000; + + private final @NotNull SentryOptions options; + private final @NotNull MonotonicTicker ticker; + private final long correlationWindowNanos; + private final int maxTrackedThrowables; + private final @NotNull Map capturedThrowables = + Collections.synchronizedMap(new WeakHashMap<>()); + + ScopeBasedDuplicateEventDetectionEventProcessor(final @NotNull SentryOptions options) { + this( + options, + options.getMonotonicTicker(), + DEFAULT_CORRELATION_WINDOW_NANOS, + DEFAULT_MAX_TRACKED_THROWABLES); + } + + ScopeBasedDuplicateEventDetectionEventProcessor( + final @NotNull SentryOptions options, + final @NotNull MonotonicTicker ticker, + final long correlationWindowNanos, + final int maxTrackedThrowables) { + this.options = options; + this.ticker = ticker; + this.correlationWindowNanos = correlationWindowNanos; + this.maxTrackedThrowables = maxTrackedThrowables; + } + + @Override + public @Nullable SentryEvent process(final @NotNull SentryEvent event, final @NotNull Hint hint) { + final @Nullable Throwable throwable = event.getThrowable(); + if (throwable == null) { + return event; + } + + final @NotNull CaptureSource source = getCaptureSource(event.getThrowableMechanism()); + final long now = ticker.tickNanos(); + boolean duplicate = false; + + synchronized (capturedThrowables) { + pruneStaleEntries(now); + final @Nullable CapturedThrowable previous = capturedThrowables.get(throwable); + if (previous == null) { + if (capturedThrowables.size() < maxTrackedThrowables) { + capturedThrowables.put(throwable, new CapturedThrowable(source, now)); + } + } else if (previous.source.isSameSource(source)) { + capturedThrowables.put(throwable, new CapturedThrowable(source, now)); + } else { + capturedThrowables.remove(throwable); + duplicate = true; + } + } + + if (duplicate) { + options + .getLogger() + .log( + SentryLevel.DEBUG, + "Duplicate Exception from a different capture source detected. Event %s will be discarded.", + event.getEventId()); + return null; + } + return event; + } + + @Override + public @NotNull EventProcessor fork() { + return new ScopeBasedDuplicateEventDetectionEventProcessor( + options, ticker, correlationWindowNanos, maxTrackedThrowables); + } + + private void pruneStaleEntries(final long now) { + final Iterator> iterator = + capturedThrowables.entrySet().iterator(); + while (iterator.hasNext()) { + final CapturedThrowable captured = iterator.next().getValue(); + if (now - captured.timestampNanos >= correlationWindowNanos) { + iterator.remove(); + } + } + } + + private static @NotNull CaptureSource getCaptureSource( + final @Nullable Throwable throwableMechanism) { + if (throwableMechanism instanceof ExceptionMechanismException) { + final @Nullable String mechanismType = + ((ExceptionMechanismException) throwableMechanism).getExceptionMechanism().getType(); + if (mechanismType != null && !mechanismType.isEmpty()) { + return new CaptureSource(mechanismType); + } + } + return CaptureSource.UNKNOWN; + } + + private static final class CapturedThrowable { + private final @NotNull CaptureSource source; + private final long timestampNanos; + + private CapturedThrowable(final @NotNull CaptureSource source, final long timestampNanos) { + this.source = source; + this.timestampNanos = timestampNanos; + } + } + + private static final class CaptureSource { + private static final @NotNull CaptureSource UNKNOWN = new CaptureSource(null); + + private final @Nullable String mechanismType; + + private CaptureSource(final @Nullable String mechanismType) { + this.mechanismType = mechanismType; + } + + private boolean isSameSource(final @NotNull CaptureSource other) { + if (mechanismType == null) { + return other.mechanismType == null; + } + return mechanismType.equals(other.mechanismType); + } + } +} diff --git a/sentry/src/test/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessorTest.kt b/sentry/src/test/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessorTest.kt new file mode 100644 index 0000000000..3b3614a216 --- /dev/null +++ b/sentry/src/test/java/io/sentry/ScopeBasedDuplicateEventDetectionEventProcessorTest.kt @@ -0,0 +1,321 @@ +package io.sentry + +import com.google.common.truth.Truth.assertThat +import io.sentry.exception.ExceptionMechanismException +import io.sentry.protocol.Mechanism +import io.sentry.time.TestMonotonicTicker +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import kotlin.test.Test + +class ScopeBasedDuplicateEventDetectionEventProcessorTest { + private val ticker = TestMonotonicTicker() + private val options = SentryOptions() + + @Test + fun `first capture sends and stores exact mechanism source`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNull() + } + + @Test + fun `same throwable and same exact mechanism source sends repeatedly`() { + val processor = processor() + val throwable = RuntimeException() + + repeat(3) { + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + } + } + + @Test + fun `different mechanism source drops once and consumes the pair`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNull() + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + } + + @Test + fun `reverse mechanism source order drops once`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNull() + } + + @Test + fun `direct throwable uses unknown source and sends repeatedly`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(SentryEvent(throwable), Hint())).isNotNull() + assertThat(processor.process(SentryEvent(throwable), Hint())).isNotNull() + } + + @Test + fun `null and empty mechanism types use the same unknown source`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, null), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, ""), Hint())).isNotNull() + } + + @Test + fun `unknown followed by concrete mechanism drops once`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(SentryEvent(throwable), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNull() + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + } + + @Test + fun `concrete mechanism followed by unknown drops once`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + assertThat(processor.process(SentryEvent(throwable), Hint())).isNull() + } + + @Test + fun `distinct logging mechanism types are different sources`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Log4j2"), Hint())).isNull() + } + + @Test + fun `events without throwables send`() { + assertThat(processor().process(SentryEvent(), Hint())).isNotNull() + } + + @Test + fun `direct mechanism wrapper is unwrapped for the map key`() { + val processor = processor() + val throwable = RuntimeException() + + val first = mechanismEvent(throwable, "Logback") + val second = mechanismEvent(throwable, "Spring") + + assertThat(first.getThrowable()).isSameInstanceAs(throwable) + assertThat(processor.process(first, Hint())).isNotNull() + assertThat(processor.process(second, Hint())).isNull() + } + + @Test + fun `shared nested causes do not deduplicate`() { + val processor = processor() + val sharedCause = RuntimeException() + val first = RuntimeException("first", sharedCause) + val second = RuntimeException("second", sharedCause) + + assertThat(processor.process(mechanismEvent(first, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(second, "Spring"), Hint())).isNotNull() + } + + @Test + fun `static stackless singleton sends for repeated same source captures`() { + val processor = processor() + val singleton = StacklessException() + + repeat(3) { + assertThat(processor.process(mechanismEvent(singleton, "ReactorNetty"), Hint())).isNotNull() + } + } + + @Test + fun `stale candidates send and are replaced`() { + val processor = processor() + val throwable = RuntimeException() + + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + ticker.advance(1, TimeUnit.SECONDS) + + assertThat(processor.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNull() + } + + @Test + fun `stale entries are pruned during later processor activity`() { + val processor = processor(maxTrackedThrowables = 1) + val staleThrowable = RuntimeException() + val currentThrowable = RuntimeException() + + assertThat(processor.process(mechanismEvent(staleThrowable, "Logback"), Hint())).isNotNull() + ticker.advance(1, TimeUnit.SECONDS) + assertThat(processor.process(mechanismEvent(currentThrowable, "Logback"), Hint())).isNotNull() + + assertThat(processor.process(mechanismEvent(currentThrowable, "Spring"), Hint())).isNull() + } + + @Test + fun `capacity exhaustion fails open for new throwable keys`() { + val processor = processor(maxTrackedThrowables = 1) + val tracked = RuntimeException() + val untracked = RuntimeException() + + assertThat(processor.process(mechanismEvent(tracked, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(untracked, "Logback"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(untracked, "Spring"), Hint())).isNotNull() + assertThat(processor.process(mechanismEvent(tracked, "Spring"), Hint())).isNull() + } + + @Test + fun `cyclic causes are irrelevant`() { + val processor = processor() + val throwable = CauseReadingFailsException() + + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + } + + @Test + fun `concurrent same-source captures all send`() { + val processor = processor() + val throwable = RuntimeException() + val executor = Executors.newFixedThreadPool(8) + val start = CountDownLatch(1) + + try { + val results = + (1..16).map { + executor.submit { + start.await() + processor.process(mechanismEvent(throwable, "Logback"), Hint()) + } + } + start.countDown() + + assertThat(results.map { it.get(5, TimeUnit.SECONDS) }).doesNotContain(null) + } finally { + executor.shutdownNow() + } + } + + @Test + fun `concurrent different-source captures atomically consume one pair`() { + val processor = processor() + val throwable = RuntimeException() + val executor = Executors.newFixedThreadPool(2) + val start = CountDownLatch(1) + + try { + val logback = + executor.submit { + start.await() + processor.process(mechanismEvent(throwable, "Logback"), Hint()) + } + val spring = + executor.submit { + start.await() + processor.process(mechanismEvent(throwable, "Spring"), Hint()) + } + start.countDown() + + assertThat(listOf(logback.get(), spring.get()).count { it == null }).isEqualTo(1) + } finally { + executor.shutdownNow() + } + } + + @Test + fun `fork preserves configuration and starts with empty state`() { + val processor = processor(maxTrackedThrowables = 1) + val throwable = RuntimeException() + assertThat(processor.process(mechanismEvent(throwable, "Logback"), Hint())).isNotNull() + + val fork = processor.fork() as ScopeBasedDuplicateEventDetectionEventProcessor + + assertThat(fork).isNotSameInstanceAs(processor) + assertThat(fork.process(mechanismEvent(throwable, "Spring"), Hint())).isNotNull() + assertThat(fork.process(mechanismEvent(throwable, "Logback"), Hint())).isNull() + } + + @Test + fun `duplicate logging occurs outside the state map lock`() { + val processedWhileLogging = AtomicBoolean(false) + val logger = BlockingCheckLogger() + val loggingOptions = + SentryOptions().apply { + isDebug = true + setLogger(logger) + } + val processor = + ScopeBasedDuplicateEventDetectionEventProcessor( + loggingOptions, + ticker, + ScopeBasedDuplicateEventDetectionEventProcessor.DEFAULT_CORRELATION_WINDOW_NANOS, + ScopeBasedDuplicateEventDetectionEventProcessor.DEFAULT_MAX_TRACKED_THROWABLES, + ) + logger.onLog = { + val thread = Thread { processor.process(SentryEvent(RuntimeException()), Hint()) } + logger.processingThread = thread + thread.start() + thread.join(1000) + processedWhileLogging.set(!thread.isAlive) + } + val throwable = RuntimeException() + processor.process(mechanismEvent(throwable, "Logback"), Hint()) + + processor.process(mechanismEvent(throwable, "Spring"), Hint()) + logger.processingThread?.join(5000) + + assertThat(processedWhileLogging.get()).isTrue() + } + + private fun processor( + maxTrackedThrowables: Int = + ScopeBasedDuplicateEventDetectionEventProcessor.DEFAULT_MAX_TRACKED_THROWABLES + ) = + ScopeBasedDuplicateEventDetectionEventProcessor( + options, + ticker, + ScopeBasedDuplicateEventDetectionEventProcessor.DEFAULT_CORRELATION_WINDOW_NANOS, + maxTrackedThrowables, + ) + + private fun mechanismEvent(throwable: Throwable, type: String?): SentryEvent { + val mechanism = Mechanism().apply { this.type = type } + return SentryEvent(ExceptionMechanismException(mechanism, throwable, Thread.currentThread())) + } + + private class StacklessException : RuntimeException(null, null, false, false) + + private class CauseReadingFailsException : RuntimeException() { + override val cause: Throwable? + get() = error("The processor must not inspect cause chains") + } + + private class BlockingCheckLogger : ILogger { + var onLog: (() -> Unit)? = null + var processingThread: Thread? = null + + override fun log(level: SentryLevel, message: String, vararg args: Any?) { + onLog?.invoke() + } + + override fun log(level: SentryLevel, message: String, throwable: Throwable?) = Unit + + override fun log( + level: SentryLevel, + throwable: Throwable?, + message: String, + vararg args: Any?, + ) = Unit + + override fun isEnabled(level: SentryLevel?): Boolean = true + } +}