diff --git a/news/6960.bugfix.md b/news/6960.bugfix.md new file mode 100644 index 00000000000..7877c535fb6 --- /dev/null +++ b/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. diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index a1d76d2e8b5..f2f8b794301 100644 --- a/reflex/utils/telemetry.py +++ b/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: diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 361a52dd537..334ef431d8d 100644 --- a/tests/units/test_config.py +++ b/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() diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index aa530e39ccf..113d01c4d05 100644 --- a/tests/units/test_telemetry.py +++ b/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")