From 7ad544fe8c98b675aad340e4e6d886c831eabf88 Mon Sep 17 00:00:00 2001 From: Erica Pisani Date: Mon, 20 Apr 2026 12:13:05 -0400 Subject: [PATCH] fix(dedupe): Avoid retaining strong references to user exceptions DedupeIntegration stored the original exception instance as a fallback when `weakref.ref(exc)` failed (e.g. for builtin exception types). That meant a reference to every such exception lived on the context var until the next exception replaced it, potentially pinning large traceback frames, locals, and user objects in memory. Instead, fall back to a `(type, hash(args))` fingerprint so dedupe can still suppress duplicates without keeping the exception alive. Refs PY-2381 Co-Authored-By: Claude Opus 4.7 --- sentry_sdk/integrations/dedupe.py | 33 ++++++++++++++++++++++--------- 1 file changed, 24 insertions(+), 9 deletions(-) diff --git a/sentry_sdk/integrations/dedupe.py b/sentry_sdk/integrations/dedupe.py index 09e60e4be6..46c80f8e68 100644 --- a/sentry_sdk/integrations/dedupe.py +++ b/sentry_sdk/integrations/dedupe.py @@ -8,11 +8,25 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from typing import Optional + from typing import Optional, Tuple, Type, Any from sentry_sdk._types import Event, Hint +def _safe_args_hash(args: "Tuple[Any, ...]") -> int: + try: + return hash(args) + except Exception: + try: + return hash(repr(args)) + except Exception: + return 0 + + +def _fingerprint(exc: BaseException) -> "Tuple[Type[BaseException], int]": + return (type(exc), _safe_args_hash(exc.args)) + + class DedupeIntegration(Integration): identifier = "dedupe" @@ -35,14 +49,15 @@ def processor(event: "Event", hint: "Optional[Hint]") -> "Optional[Event]": return event last_seen = integration._last_seen.get(None) - if last_seen is not None: - # last_seen is either a weakref or the original instance - last_seen = ( - last_seen() if isinstance(last_seen, weakref.ref) else last_seen - ) - exc = exc_info[1] - if last_seen is exc: + + is_duplicate = False + if isinstance(last_seen, weakref.ref): + is_duplicate = last_seen() is exc + elif isinstance(last_seen, tuple): + is_duplicate = last_seen == _fingerprint(exc) + + if is_duplicate: logger.info("DedupeIntegration dropped duplicated error event %s", exc) return None @@ -50,7 +65,7 @@ def processor(event: "Event", hint: "Optional[Hint]") -> "Optional[Event]": try: integration._last_seen.set(weakref.ref(exc)) except TypeError: - integration._last_seen.set(exc) + integration._last_seen.set(_fingerprint(exc)) return event