fix(batcher): Reset lock and flusher in child after fork

Uses the same fix introduced in #6148 to prevent deadlocks in the
monitor when os.fork() is called while another thread holds the
monitor's lock.

If os.fork() runs while another thread holds Batcher._lock, the child
inherits the lock locked but the holding thread does not exist in the
child, so the lock can never be released and _ensure_thread deadlocks
forever. Register an after-fork hook via os.register_at_fork that
replaces _lock with a fresh lock and resets _flusher / _flusher_pid
in the child. Use a weakref to the batcher so the hook does not keep
the instance alive.

Move shared init out of SpanBatcher into the base Batcher.__init__ so
all batchers (log, metrics, span) get the fork-safety hook from a
single place.

Fixes PY-2391
Fixes #6149

Author: ericapisani

sentry_sdk/_batcher.py

@@ -3,6 +3,7 @@
 import threading
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, TypeVar, Generic
+import weakref
 
 from sentry_sdk.utils import format_timestamp
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
@@ -38,6 +39,24 @@ def __init__(
         self._flusher: "Optional[threading.Thread]" = None
         self._flusher_pid: "Optional[int]" = None
 
+        self_ref = weakref.ref(self)
+
+        def _reset_thread_state() -> None:
+            batcher = self_ref()
+
+            if batcher is not None:
+                batcher._flusher = None
+                batcher._lock = threading.Lock()
+                batcher._flusher_pid = None
+
+        # Same as https://github.com/getsentry/sentry-python/issues/6148.
+        # If os.fork() runs while another thread holds self._lock,
+        # the child inherits the lock locked but the holding thread does
+        # not exist in the child, so the lock can never be released and
+        # _ensure_thread deadlocks forever.
+        if hasattr(os, "register_at_fork"):
+            os.register_at_fork(after_in_child=_reset_thread_state)
+
     def _ensure_thread(self) -> bool:
         """For forking processes we might need to restart this thread.
         This ensures that our process actually has that thread running.

sentry_sdk/_span_batcher.py

@@ -1,4 +1,3 @@
-import threading
 from collections import defaultdict
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
@@ -8,7 +7,7 @@
 from sentry_sdk.utils import format_timestamp, serialize_attribute
 
 if TYPE_CHECKING:
-    from typing import Any, Callable, Optional
+    from typing import Any, Callable
     from sentry_sdk.traces import StreamedSpan
 
 
@@ -32,23 +31,14 @@ def __init__(
         capture_func: "Callable[[Envelope], None]",
         record_lost_func: "Callable[..., None]",
     ) -> None:
+        super().__init__(capture_func, record_lost_func)
         # Spans from different traces cannot be emitted in the same envelope
         # since the envelope contains a shared trace header. That's why we bucket
         # by trace_id, so that we can then send the buckets each in its own
         # envelope.
         # trace_id -> span buffer
         self._span_buffer: dict[str, list["StreamedSpan"]] = defaultdict(list)
         self._running_size: dict[str, int] = defaultdict(lambda: 0)
-        self._capture_func = capture_func
-        self._record_lost_func = record_lost_func
-        self._running = True
-        self._lock = threading.Lock()
-        self._active: "threading.local" = threading.local()
-
-        self._flush_event: "threading.Event" = threading.Event()
-
-        self._flusher: "Optional[threading.Thread]" = None
-        self._flusher_pid: "Optional[int]" = None
 
     def add(self, span: "StreamedSpan") -> None:
         # Bail out if the current thread is already executing batcher code.

tests/test_logs.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import os
 import sys
 import time
 from typing import List, Any, Mapping, Union
@@ -819,3 +820,40 @@ def add_to_envelope_with_reentrant_add(envelope):
     assert reentrant_add_called
     # If the re-entrancy guard didn't work, this test would hang and it'd
     # eventually be timed out by pytest-timeout
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32"
+    or not hasattr(os, "fork")
+    or not hasattr(os, "register_at_fork"),
+    reason="requires POSIX fork and os.register_at_fork (Python 3.7+)",
+)
+def test_log_batcher_lock_reset_in_child_after_fork(sentry_init):
+    """Regression test for the LogBatcher fork-deadlock fix.
+
+    If os.fork() runs while another thread holds LogBatcher._lock, the
+    child inherits the lock locked. The holding thread does not exist in
+    the child, so the lock can never be released and _ensure_thread
+    deadlocks forever. The after-fork hook must replace the lock with a
+    fresh one in the child and reset _flusher / _flusher_pid.
+    """
+    sentry_init(enable_logs=True)
+    batcher = sentry_sdk.get_client().log_batcher
+    assert batcher is not None
+
+    original_lock = batcher._lock
+    original_lock.acquire()
+    pid = os.fork()
+    if pid == 0:
+        # Child: was the lock object replaced and is the new one not
+        # held? Without the fix, _lock is `original_lock` inherited
+        # locked, so `replaced` is False. blocking=False guarantees the
+        # child can't hang on a regression.
+        replaced = batcher._lock is not original_lock
+        unheld = batcher._lock.acquire(blocking=False)
+        flusher_reset = batcher._flusher is None and batcher._flusher_pid is None
+        os._exit(0 if replaced and unheld and flusher_reset else 1)
+
+    original_lock.release()
+    _, status = os.waitpid(pid, 0)
+    assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0

tests/test_metrics.py

@@ -1,6 +1,9 @@
+import os
+import sys
 from typing import List
 from unittest import mock
 
+import pytest
 
 import sentry_sdk
 from sentry_sdk import get_client
@@ -512,3 +515,41 @@ def before_send_metric(metric, _):
     )
 
     get_client().flush()
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32"
+    or not hasattr(os, "fork")
+    or not hasattr(os, "register_at_fork"),
+    reason="requires POSIX fork and os.register_at_fork (Python 3.7+)",
+)
+def test_metrics_batcher_lock_reset_in_child_after_fork(sentry_init):
+    """Regression test for the MetricsBatcher fork-deadlock fix.
+
+    If os.fork() runs while another thread holds MetricsBatcher._lock,
+    the child inherits the lock locked. The holding thread does not
+    exist in the child, so the lock can never be released and
+    _ensure_thread deadlocks forever. The after-fork hook must replace
+    the lock with a fresh one in the child and reset
+    _flusher / _flusher_pid.
+    """
+    sentry_init()
+    batcher = sentry_sdk.get_client().metrics_batcher
+    assert batcher is not None
+
+    original_lock = batcher._lock
+    original_lock.acquire()
+    pid = os.fork()
+    if pid == 0:
+        # Child: was the lock object replaced and is the new one not
+        # held? Without the fix, _lock is `original_lock` inherited
+        # locked, so `replaced` is False. blocking=False guarantees the
+        # child can't hang on a regression.
+        replaced = batcher._lock is not original_lock
+        unheld = batcher._lock.acquire(blocking=False)
+        flusher_reset = batcher._flusher is None and batcher._flusher_pid is None
+        os._exit(0 if replaced and unheld and flusher_reset else 1)
+
+    original_lock.release()
+    _, status = os.waitpid(pid, 0)
+    assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0

tests/tracing/test_span_streaming.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import re
 import sys
 import time
@@ -1593,3 +1594,43 @@ def test_transport_format(sentry_init, capture_envelopes):
         assert "value" in value
         assert "type" in value
         assert value["type"] in ("string", "boolean", "integer", "double", "array")
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32"
+    or not hasattr(os, "fork")
+    or not hasattr(os, "register_at_fork"),
+    reason="requires POSIX fork and os.register_at_fork (Python 3.7+)",
+)
+def test_span_batcher_lock_reset_in_child_after_fork(sentry_init):
+    """Regression test for the SpanBatcher fork-deadlock fix.
+
+    If os.fork() runs while another thread holds SpanBatcher._lock, the
+    child inherits the lock locked. The holding thread does not exist in
+    the child, so the lock can never be released and _ensure_thread
+    deadlocks forever. The after-fork hook must replace the lock with a
+    fresh one in the child and reset _flusher / _flusher_pid.
+    """
+    sentry_init(
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"},
+    )
+    batcher = sentry_sdk.get_client().span_batcher
+    assert batcher is not None
+
+    original_lock = batcher._lock
+    original_lock.acquire()
+    pid = os.fork()
+    if pid == 0:
+        # Child: was the lock object replaced and is the new one not
+        # held? Without the fix, _lock is `original_lock` inherited
+        # locked, so `replaced` is False. blocking=False guarantees the
+        # child can't hang on a regression.
+        replaced = batcher._lock is not original_lock
+        unheld = batcher._lock.acquire(blocking=False)
+        flusher_reset = batcher._flusher is None and batcher._flusher_pid is None
+        os._exit(0 if replaced and unheld and flusher_reset else 1)
+
+    original_lock.release()
+    _, status = os.waitpid(pid, 0)
+    assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0