Preserve RegistrationContext in telemetry worker threads

news/6960.bugfix.md

@@ -0,0 +1 @@
+Telemetry events are now collected under the submitting thread's registration context, so the background worker reuses the config the app already loaded instead of re-importing `rxconfig.py` (and mutating `sys.path`) off-thread.

reflex/utils/telemetry.py

@@ -22,6 +22,7 @@
 from reflex_base import constants
 from reflex_base.config import get_config
 from reflex_base.environment import environment
+from reflex_base.registry import RegistrationContext
 from reflex_base.utils.decorator import once, once_unless_none
 from reflex_base.utils.exceptions import ReflexError
 from typing_extensions import NotRequired
@@ -495,33 +496,80 @@ def _get_telemetry_executor() -> ThreadPoolExecutor:
     return _executor
 
 
-def _run_suppressed(fn: Callable[..., Any], /, *args, **kwargs) -> None:
+def _current_registration_context() -> RegistrationContext | None:
+    """Return the caller's RegistrationContext, or None if none is attached.
+
+    Unlike ``ensure_context()`` this never attaches a context to the caller: a
+    thread that has none keeps none.
+
+    Returns:
+        The attached RegistrationContext, or None.
+    """
+    try:
+        return RegistrationContext.get()
+    except LookupError:
+        return None
+
+
+def _run_suppressed(
+    registration_context: RegistrationContext | None,
+    fn: Callable[..., Any],
+    /,
+    *args,
+    **kwargs,
+) -> None:
     """Run ``fn`` in the worker thread, never letting a failure escape.
 
+    The submitting thread's RegistrationContext is attached for the duration of
+    the call, so the config the app already loaded is reused instead of the
+    worker importing ``rxconfig.py`` again into a context of its own.
+
     Telemetry must never break the app, so any error (including a failed send)
     is reported at debug level and otherwise discarded.
 
+    Caveat: a job submitted before any context exists runs without one, so a
+    config lookup inside it attaches a context of the worker's own that later
+    context-less jobs then reuse. Harmless for a Reflex app (one app, one config
+    per process), and once the app's context has loaded its config, every
+    subsequent send carries that context in and overrides the worker's.
+
     Args:
+        registration_context: The submitter's RegistrationContext, or None when
+            it had none attached.
         fn: The callable to run.
         args: Positional arguments forwarded to ``fn``.
         kwargs: Keyword arguments forwarded to ``fn``.
     """
+    token = (
+        None
+        if registration_context is None
+        else RegistrationContext.set(registration_context)
+    )
     try:
         fn(*args, **kwargs)
     except Exception as err:
         logger.debug(f"Failed to process telemetry event: {err}")
+    finally:
+        if token is not None:
+            RegistrationContext.reset(token)
 
 
 def _submit(fn: Callable[..., Any], /, *args, **kwargs) -> None:
     """Queue telemetry work on the background executor, swallowing all errors.
 
+    The caller's RegistrationContext travels with the job so event collection
+    sees the same config (and registrations) the caller does.
+
     Args:
         fn: The callable to run in the telemetry worker thread.
         args: Positional arguments forwarded to ``fn``.
         kwargs: Keyword arguments forwarded to ``fn``.
     """
+    registration_context = _current_registration_context()
     with suppress(Exception):
-        _get_telemetry_executor().submit(_run_suppressed, fn, *args, **kwargs)
+        _get_telemetry_executor().submit(
+            _run_suppressed, registration_context, fn, *args, **kwargs
+        )
 
 
 def _flush(timeout: float | None = None) -> bool:

tests/units/test_config.py

@@ -862,11 +862,17 @@ def test_get_config_loads_once_for_shared_context(monkeypatch: pytest.MonkeyPatc
     n_threads = 8
     load_count = 0
     count_lock = threading.Lock()
+    # Only count loads made by this test's worker threads: unrelated background
+    # threads (e.g. the telemetry worker, which has no RegistrationContext of its
+    # own) may call get_config() while the patch below is installed and would
+    # otherwise be counted as a duplicate load of the shared context.
+    under_test = threading.local()
 
     def slow_load() -> rx.Config:
         nonlocal load_count
-        with count_lock:
-            load_count += 1
+        if getattr(under_test, "active", False):
+            with count_lock:
+                load_count += 1
         # Widen the check-to-set window so an unserialized load path races.
         time.sleep(0.05)
         return rx.Config(app_name="shared")
@@ -878,6 +884,7 @@ def slow_load() -> rx.Config:
     results: list[rx.Config | None] = [None] * n_threads
 
     def worker(i: int) -> None:
+        under_test.active = True
         RegistrationContext._context_var.set(ctx)
         barrier.wait()
         results[i] = reflex_base.config.get_config()

tests/units/test_telemetry.py

@@ -7,7 +7,10 @@
 import pytest
 from packaging.version import parse as parse_python_version
 from pytest_mock import MockerFixture
+from reflex_base.config import get_config
+from reflex_base.registry import RegistrationContext
 
+import reflex as rx
 from reflex.utils import telemetry
 
 
@@ -710,6 +713,59 @@ def record(*_args, **_kwargs) -> bool:
     assert seen["thread"] is not loop_thread
 
 
+def test_submit_runs_job_in_callers_registration_context():
+    """Queued telemetry work runs under the submitting thread's context.
+
+    The worker thread carries no RegistrationContext of its own, so a config
+    lookup during event collection (e.g. ``get_bun_path``) used to attach a
+    throwaway context and re-import ``rxconfig.py`` off-thread. The submitter's
+    context travels with the job instead, so the already-loaded config is reused.
+    """
+    seen: dict[str, object] = {}
+
+    def record() -> None:
+        seen["thread"] = threading.current_thread()
+        seen["context"] = RegistrationContext.get()
+        seen["config"] = get_config()
+
+    with RegistrationContext() as ctx:
+        config = rx.Config(app_name="telemetry_ctx")
+        ctx._set_config(config)
+
+        telemetry._submit(record)
+        telemetry._flush()
+
+    assert seen["thread"] is not threading.current_thread()
+    assert seen["context"] is ctx
+    assert seen["config"] is config
+
+
+def test_submit_leaves_worker_context_clean_between_jobs():
+    """The attached context is detached again once the job finishes."""
+    seen: list[RegistrationContext | None] = []
+
+    def record() -> None:
+        try:
+            seen.append(RegistrationContext.get())
+        except LookupError:
+            seen.append(None)
+
+    with RegistrationContext() as ctx:
+        ctx._set_config(rx.Config(app_name="telemetry_ctx"))
+        telemetry._submit(record)
+        telemetry._flush()
+
+    # Submitted from a thread that is not under ``ctx``: the worker must not
+    # still be holding the previous job's context.
+    submitter = threading.Thread(target=lambda: telemetry._submit(record))
+    submitter.start()
+    submitter.join()
+    telemetry._flush()
+
+    assert seen[0] is ctx
+    assert seen[1] is not ctx
+
+
 def test_send_suppresses_worker_errors(mocker: MockerFixture):
     """A failed telemetry send is swallowed and never reaches the caller."""
     mocker.patch.object(telemetry, "_maybe_alias_legacy_distinct_id")