bukzor/di 1163 thread leak detection dark launch (#97320)

- **DI-1008: assert that tests don't leak threads**
- **remove singletonproduer._shutdown_all() hack**
- **custom diff, for unambiguous +/- thread blocks**
- **attempt to keep only test_ (or testutils) frames, if we have them**
- **pytest config: pytest.mark.thread_leak_allowlist**
- **DI-1163: dark-launch thread-leak assertions**
- **remove "hax"**
- **test for thread-leak differ**

Author: bukzor

pyproject.toml

@@ -217,6 +217,7 @@ markers = [
   "sentry_metrics: test requires access to sentry metrics",
   "symbolicator: test requires access to symbolicator",
   "querybuilder: smoke tests for QueryBuilders",
+  "thread_leak_allowlist: temporarily allow known-issue thread leaks",
 ]
 filterwarnings = [
   # Consider all warnings to be errors other than the ignored ones.
@@ -583,6 +584,7 @@ module = [
     "sentry.testutils.pytest.json_report_reruns",
     "sentry.testutils.pytest.show_flaky_failures",
     "sentry.testutils.skips",
+    "sentry.testutils.thread_leaks.*",
     "sentry.toolbar.utils.*",
     "sentry.types.*",
     "sentry.uptime.migrations.*",
@@ -774,6 +776,7 @@ module = [
     "tests.sentry.tasks.test_delete_seer_grouping_records",
     "tests.sentry.tasks.test_on_demand_metrics",
     "tests.sentry.testutils.helpers.*",
+    "tests.sentry.testutils.thread_leaks.*",
     "tests.sentry.types.*",
     "tests.sentry.usage_accountant.*",
     "tests.sentry.users.api.bases.*",

src/sentry/testutils/thread_leaks/__init__.py

@@ -0,0 +1,66 @@
+"""Utility used to regression-proof our thread-leak fixes.
+
+It turns out due to various indirections, just showing the name and exact
+"target" of a thread isn't enough to find where/how it was spawned. This code
+does a bit of work to record and show exactly where the Thread came from, which
+proved essential when working to fix these things.
+"""
+
+import os
+import threading
+import traceback
+from collections.abc import Generator
+from contextlib import contextmanager
+from threading import Thread
+from traceback import StackSummary
+from typing import Any
+from unittest import mock
+
+from . import sentry
+from .diff import diff
+
+_cwd = os.getcwd() + "/"
+del os  # hygiene
+
+
+class ThreadLeakAssertionError(AssertionError):
+    pass
+
+
+def _where(cwd: str = _cwd) -> StackSummary:
+    stack = traceback.extract_stack()
+    for frame in stack:
+        frame.filename = frame.filename.replace(cwd, "./")  # for readability
+    return stack
+
+
+@contextmanager
+def threading_remembers_where() -> Generator[None]:
+    """Smuggle a ._where StackSummary attribute onto each Thread construction."""
+    __init__ = Thread.__init__
+
+    def patched__init__(self: Thread, *a: Any, **k: Any) -> None:
+        setattr(self, "_where", _where())
+        __init__(self, *a, **k)
+
+    with mock.patch.object(Thread, "__init__", patched__init__):
+        yield
+
+
+@contextmanager
+def assert_none(strict: bool = True) -> Generator[dict[str, Any]]:
+    """Assert no thread leaks occurred during context execution."""
+
+    with threading_remembers_where():
+        expected = threading.enumerate()
+        result: dict[str, Any] = {"events": {}}
+        yield result
+        actual = threading.enumerate()
+
+        thread_leaks = set(actual) - set(expected)
+        if not thread_leaks:
+            return
+
+        result["events"] = sentry.capture_event(thread_leaks, strict)
+        if strict:
+            raise ThreadLeakAssertionError(diff(old=expected, new=actual))

src/sentry/testutils/thread_leaks/assertion.py

@@ -0,0 +1,99 @@
+"""Generate readable diffs showing thread creation locations for debugging leaks."""
+
+import sys
+from collections.abc import Generator, Iterable
+from threading import Thread
+from traceback import FrameSummary, StackSummary
+
+
+def get_relevant_frames(stack: Iterable[FrameSummary]) -> StackSummary:
+    """Filter stack frames to show only test and testutil code relevant for debugging.
+
+    Applies a series of filters to remove stdlib and system code, then focuses on
+    test and testutil frames where thread leak fixes typically need to go. Each filter
+    is only applied if it doesn't remove all frames.
+    """
+    # Get stdlib directory path if available (for filtering system code)
+    stdlib_dir = getattr(sys, "_stdlib_dir", None)
+
+    # Apply each filter independently, reverting if it would remove all frames
+    for filter in (
+        lambda frame: frame.filename == __file__,  # Remove this very file
+        lambda frame: frame.line is None,  # Remove "frozen" stdlib modules
+        lambda frame: stdlib_dir
+        and frame.filename.startswith(stdlib_dir + "/"),  # Remove Python stdlib
+        lambda frame: frame.filename.startswith(sys.prefix),  # Remove system Python files
+        # Keep only test/testutil frames (where fixes will need to go)
+        lambda frame: "/test_" not in frame.filename,  # NOTE: filter-not is a double negative
+        lambda frame: "/testutils/" not in frame.filename,
+    ):
+        filtered_stack = [frame for frame in stack if not filter(frame)]
+        # Only apply the filter if it leaves some frames (fail-safe)
+        if filtered_stack:
+            stack = StackSummary.from_list(filtered_stack)
+
+    return StackSummary.from_list(stack)
+
+
+def _get_thread_function_name(thread: Thread) -> str:
+    """Extract fully qualified function name from thread target.
+
+    Handles cases where thread target is None or wrapped (e.g., functools.partial).
+    Returns a string representation suitable for debugging output.
+    """
+    func = getattr(thread, "_target", None)
+    if func is None:
+        return "None"
+
+    # Use __qualname__ if available, fallback to str() for complex objects like functools.partial
+    func_name = getattr(func, "__qualname__", str(func))
+    return f"{func.__module__}.{func_name}"
+
+
+def _threads_to_diffable(threads: list[Thread]) -> list[str]:
+    """Convert threads to string representations suitable for diffing.
+
+    Each thread becomes a formatted string containing:
+    - Thread representation (name, ID, daemon status)
+    - Target function fully qualified name
+    - Indented stack trace showing where thread was created
+
+    Threads are sorted by ID for consistent diff output.
+    """
+    result: list[str] = []
+    for thread in sorted(threads, key=lambda t: t.ident or 0):
+        func_fqname = _get_thread_function_name(thread)
+        stack = getattr(thread, "_where", [])
+        stack = get_relevant_frames(stack)
+        stack = "".join(stack.format())
+        # Indent stack trace lines for visual hierarchy in diff output
+        stack = ("\n" + stack).replace("\n", "\n  ").strip(" ")
+        result.append(f"{thread!r}@{func_fqname}{stack}\n")
+    return result
+
+
+def _diff(old: list[str], new: list[str]) -> Generator[str]:
+    """Generate unified diff lines between two thread lists."""
+    import difflib
+
+    matcher = difflib.SequenceMatcher(None, old, new)
+    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+        for tags, threads, prefix in [
+            (("equal",), old[i1:i2], " "),
+            (("delete", "replace"), old[i1:i2], "-"),
+            (("insert", "replace"), new[j1:j2], "+"),
+        ]:
+            if tag in tags:
+                for thread_str in threads:
+                    for line in thread_str.splitlines():
+                        yield f"{prefix} {line}\n"
+
+
+def diff(old: list[Thread], new: list[Thread]) -> str:
+    """Generate unambiguous unified diff from structured thread data."""
+    return "\n" + "".join(
+        _diff(
+            _threads_to_diffable(old),
+            _threads_to_diffable(new),
+        )
+    )

src/sentry/testutils/thread_leaks/diff.py

@@ -0,0 +1,28 @@
+"""Pytest integration for thread leak detection."""
+
+from collections.abc import Generator
+
+import pytest
+
+from .assertion import assert_none
+
+
+def check_test(request: pytest.FixtureRequest, strict: bool = True) -> Generator[None]:
+    """Check test for thread leaks, respecting allowlist markers."""
+    if request.node.get_closest_marker("thread_leak_allowlist"):
+        yield
+    else:
+        with assert_none(strict):
+            yield
+
+
+def allowlist(reason: str | None = None, *, issue: int) -> pytest.MarkDecorator:
+    """Mark test as allowed to leak threads with tracking issue."""
+    decorator = pytest.mark.thread_leak_allowlist(reason=reason, issue=issue)
+    return decorator
+
+
+def singleton_cleanup(reason: str | None = None, *, issue: int) -> pytest.MarkDecorator:
+    """Mark test as needing singleton cleanup with tracking issue."""
+    decorator = pytest.mark.thread_leak_singleton_cleanup(reason=reason, issue=issue)
+    return decorator

src/sentry/testutils/thread_leaks/pytest.py

@@ -0,0 +1,105 @@
+"""Sentry integration for reporting thread leak events with proper context."""
+
+from __future__ import annotations
+
+import functools
+from collections.abc import Iterable
+from threading import Thread
+from traceback import FrameSummary
+from typing import TYPE_CHECKING
+
+import sentry_sdk.scope
+
+from .diff import get_relevant_frames
+
+if TYPE_CHECKING:
+    # this is *only* defined when TYPE_CHECKING =(
+    from sentry_sdk._types import Event as SentryEvent
+
+
+@functools.cache
+def get_scope() -> sentry_sdk.scope.Scope:
+    """Create configured Sentry scope for thread leak reporting."""
+    from os import environ
+
+    sha: str | None
+    branch: str | None
+    if environ.get("GITHUB_ACTIONS") == "true":
+        sha = environ["GITHUB_SHA"]
+        branch = environ.get("GITHUB_HEAD_REF")
+        if branch:
+            environment = "PR"
+        else:
+            environment = "master"
+            branch = environ["GITHUB_REF_NAME"]
+    else:
+        sha = branch = None
+        environment = "local"
+
+    client = sentry_sdk.Client(
+        # proj-thread-leaks
+        dsn="https://[email protected]/4509798820085760",
+        environment=environment,
+    )
+
+    scope = sentry_sdk.get_current_scope().fork()
+    scope.set_client(client)
+    scope.update_from_kwargs(
+        # Don't set level - scope overrides event-level
+        extras={"git-branch": branch, "git-sha": sha},
+    )
+    return scope
+
+
+def capture_event(thread_leaks: set[Thread], strict: bool) -> dict[str, SentryEvent]:
+    """Report thread leaks to Sentry with proper event formatting."""
+    # Report to Sentry
+    scope = get_scope()
+    events = {}
+    with sentry_sdk.scope.use_scope(scope):
+        for thread_leak in thread_leaks:
+            event = get_thread_leak_event(thread_leak, strict)
+            event_id = scope.capture_event(event)
+            if event_id is not None:
+                events[event_id] = event
+        scope.client.flush()
+    return events
+
+
+def get_thread_leak_event(thread: Thread, strict: bool = True) -> SentryEvent:
+    """Create Sentry event from leaked thread."""
+    stack: Iterable[FrameSummary] = getattr(thread, "_where", [])
+    return event_from_stack(repr(thread), stack, strict)
+
+
+def event_from_stack(value: str, stack: Iterable[FrameSummary], strict: bool) -> SentryEvent:
+    relevant_frames = get_relevant_frames(stack)
+
+    # https://develop.sentry.dev/sdk/data-model/event-payloads/exception/
+    exception = {
+        "mechanism": {
+            "type": __name__,
+            "handled": not strict,
+            "help_link": "https://www.notion.so/sentry/How-To-Thread-Leaks-2488b10e4b5d8049965cc057b5fb5f6b",
+        },
+        "type": "ThreadLeakAssertionError",
+        "value": value,
+        "stacktrace": {
+            "frames": [
+                {
+                    "filename": frame.filename,
+                    "function": frame.name,
+                    "module": frame.locals.get("__name__") if frame.locals else None,
+                    "lineno": frame.lineno,
+                    "context_line": frame.line,
+                    "in_app": frame in relevant_frames,
+                }
+                for frame in stack
+            ]
+        },
+    }
+    return {
+        "level": "error" if strict else "warning",
+        "message": "Thread leak detected",
+        "exception": {"values": [exception]},
+    }

src/sentry/testutils/thread_leaks/sentry.py

@@ -11,6 +11,7 @@
 
 from sentry.silo.base import SiloMode
 from sentry.testutils.pytest.sentry import get_default_silo_mode_for_test_cases
+from sentry.testutils.thread_leaks import pytest as thread_leaks
 
 pytest_plugins = ["sentry.testutils.pytest"]
 
@@ -56,6 +57,12 @@ def unclosed_files():
     assert _open_files() == fds
 
 
+@pytest.fixture(autouse=True)
+def unclosed_threads(request):
+    # TODO(DI-1067): strict mode
+    yield from thread_leaks.check_test(request, strict=False)
+
+
 @pytest.fixture(autouse=True)
 def validate_silo_mode():
     # NOTE!  Hybrid cloud uses many mechanisms to simulate multiple different configurations of the application

tests/conftest.py

@@ -0,0 +1,19 @@
+"""Test configuration for thread leak tests."""
+
+from collections.abc import Generator
+
+import pytest
+
+
+@pytest.fixture(scope="package", autouse=True)
+def thread_leak_test_environment() -> Generator[None]:
+    """Set sentry environment to "selftest" for all tests in this package."""
+    from sentry.testutils.thread_leaks.sentry import get_scope
+
+    scope = get_scope()
+    orig = scope.client.options.get("environment")
+    scope.client.options["environment"] = "selftest"
+
+    yield
+
+    scope.client.options["environment"] = orig