fix(profiler): reset ThreadContinuousScheduler lock after fork

If os.fork() runs while ThreadContinuousScheduler.lock is held, the child
inherits a locked lock with no owner and ensure_running can deadlock.
Mirror the Monitor #6159 pattern: register_at_fork after_in_child reset.

Fixes #6165

Author: BetterAndBetterII

sentry_sdk/profiler/continuous_profiler.py

@@ -6,6 +6,7 @@
 import time
 import uuid
 import warnings
+import weakref
 from collections import deque
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
@@ -443,6 +444,8 @@ class ThreadContinuousScheduler(ContinuousScheduler):
     mode: "ContinuousProfilerMode" = "thread"
     name = "sentry.profiler.ThreadContinuousScheduler"
 
+    thread: "Optional[threading.Thread]"
+
     def __init__(
         self,
         frequency: int,
@@ -452,8 +455,34 @@ def __init__(
     ) -> None:
         super().__init__(frequency, options, sdk_info, capture_func)
 
-        self.thread: "Optional[threading.Thread]" = None
+        self._reset_thread_state()
+
+        # See https://github.com/getsentry/sentry-python/issues/6165.
+        # 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_running deadlocks forever. Reinitialise the lock,
+        # cached thread/pid, running flag, and buffer in the child so
+        # it starts clean regardless of inherited state. We bind via a
+        # WeakMethod so the permanently-registered fork handler does
+        # not pin this scheduler: register_at_fork has no unregister
+        # API. POSIX-only; Windows uses spawn.
+        if hasattr(os, "register_at_fork"):
+            weak_reset = weakref.WeakMethod(self._reset_thread_state)
+
+            def _reset_in_child() -> None:
+                method = weak_reset()
+                if method is not None:
+                    method()
+
+            os.register_at_fork(after_in_child=_reset_in_child)
+
+    def _reset_thread_state(self) -> None:
+        self.thread = None
         self.lock = threading.Lock()
+        self.pid = None
+        self.running = False
+        self.buffer = None
 
     def ensure_running(self) -> None:
         self.soft_shutdown = False

tests/profiler/test_continuous_profiler.py

@@ -1,3 +1,5 @@
+import os
+import sys
 import threading
 import time
 from collections import defaultdict
@@ -8,6 +10,7 @@
 import sentry_sdk
 from sentry_sdk.consts import VERSION
 from sentry_sdk.profiler.continuous_profiler import (
+    ThreadContinuousScheduler,
     get_profiler_id,
     is_profile_session_sampled,
     setup_continuous_profiler,
@@ -1257,3 +1260,64 @@ def test_continuous_profiler_run_does_not_null_buffer_span_streaming(
         "run() must not set self.buffer = None; "
         "this would destroy buffers created by concurrent ensure_running() calls"
     )
+
+
+@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_thread_continuous_scheduler_lock_reset_in_child_after_fork():
+    """Regression test for #6165.
+
+    If os.fork() runs while ThreadContinuousScheduler.lock is held, the
+    child inherits the lock locked. The holding thread does not exist in
+    the child, so the lock can never be released and ensure_running
+    deadlocks forever. The after-fork hook must replace the lock with a
+    fresh one in the child, and reset the inherited thread, pid, and
+    buffer so the child starts clean.
+    """
+    scheduler = ThreadContinuousScheduler(
+        frequency=101,
+        options={"profile_lifecycle": "manual", "profile_session_sample_rate": 1.0},
+        sdk_info=mock_sdk_info,
+        capture_func=lambda envelope: None,
+    )
+    scheduler.reset_buffer()
+    scheduler.thread = threading.current_thread()
+    scheduler.pid = os.getpid()
+    scheduler.running = True
+
+    original_lock = scheduler.lock
+    original_buffer = scheduler.buffer
+    assert original_buffer is not None
+
+    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 = scheduler.lock is not original_lock
+        unheld = scheduler.lock.acquire(blocking=False)
+        thread_reset = scheduler.thread is None
+        pid_reset = scheduler.pid is None
+        buffer_reset = scheduler.buffer is None
+        running_reset = scheduler.running is False
+        os._exit(
+            0
+            if replaced
+            and unheld
+            and thread_reset
+            and pid_reset
+            and buffer_reset
+            and running_reset
+            else 1
+        )
+
+    original_lock.release()
+    _, status = os.waitpid(pid, 0)
+    assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0
+