diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index 92d7e67567f..c69d1361fa0 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -25,6 +25,19 @@ config = rx.Config( See the [config reference](/docs/api-reference/config/) for all the parameters available. +## Event Loop Blocking Detection + +In development, Reflex warns when an event handler blocks the asyncio event loop near or beyond `event_loop_blocking_threshold` seconds (default `0.2`). The warning appears as soon as the threshold is reached, so a handler that never returns is reported too. It names the handler and the innermost line of your app's code in the blocking stack, then shows the full stack. A blocking task that processes no event, such as a task your app starts, is reported by its task name. When the loop responds again, a second line reports the total duration, which includes the state updates after the handler, such as computed vars. It recommends moving blocking I/O to a thread pool or using a non-blocking async API. Background events still run on the event loop and do not make synchronous blocking code non-blocking. Detection is always off in production. Set the threshold to `0` to turn it off in development: + +```python +config = rx.Config( + app_name="my_app_name", + event_loop_blocking_threshold=0, +) +``` + +The setting can also be overridden with `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD`, for example `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD=0` to turn detection off. + ## Environment Variables Any config parameter can be overridden by setting an environment variable with the `REFLEX_` prefix and the parameter name in uppercase. Environment variables take precedence over values set in `rxconfig.py`. diff --git a/news/+7619.feature.md b/news/+7619.feature.md new file mode 100644 index 00000000000..eeb9c33a28b --- /dev/null +++ b/news/+7619.feature.md @@ -0,0 +1 @@ +Add development-time detection for event handlers that block the asyncio event loop. diff --git a/packages/reflex-base/news/+7619.feature.md b/packages/reflex-base/news/+7619.feature.md new file mode 100644 index 00000000000..8e05e63ca30 --- /dev/null +++ b/packages/reflex-base/news/+7619.feature.md @@ -0,0 +1 @@ +Add configurable development-time warnings for event handlers that block the asyncio event loop. diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index c3861dc4c3d..524fb4b3098 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/packages/reflex-base/src/reflex_base/config.py @@ -3,6 +3,7 @@ import dataclasses import importlib import logging +import math import os import sys import threading @@ -194,6 +195,7 @@ class BaseConfig: plugins: List of plugins to use in the app. disable_plugins: List of plugin types to disable in the app. transport: The transport method for client-server communication. + event_loop_blocking_threshold: Minimum event-loop blocking duration in seconds to report in development. Set to 0 to disable detection. """ app_name: str @@ -285,6 +287,8 @@ class BaseConfig: transport: Literal["websocket", "polling"] = "websocket" + event_loop_blocking_threshold: float = 0.2 + # Whether to skip plugin checks. _skip_plugins_checks: bool = dataclasses.field(default=False, repr=False) @@ -399,6 +403,15 @@ def _post_init(self, **kwargs): msg = "trusted_proxy_hops must be a non-negative integer." raise ConfigError(msg) + threshold = self.event_loop_blocking_threshold + if ( + isinstance(threshold, bool) + or not isinstance(threshold, (int, float)) + or not 0 <= threshold < math.inf + ): + msg = "event_loop_blocking_threshold must be a non-negative finite number." + raise ConfigError(msg) + # Normalize route prefixes to ensure they start with a slash. self._normalize_paths() diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py new file mode 100644 index 00000000000..840d3fe149b --- /dev/null +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -0,0 +1,186 @@ +"""Development-time detection of tasks blocking the asyncio loop.""" + +from __future__ import annotations + +import asyncio +import logging +import os +import sys +import sysconfig +import threading +import time +import traceback +from collections.abc import Mapping +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from reflex_base.registry import RegisteredEventHandler + +logger = logging.getLogger(__name__) + +# Name prefix of the asyncio task that processes one event. +EVENT_TASK_PREFIX = "reflex_event|" + +# The standard library and installed packages hold no app code. +_LIBRARY_PATHS = tuple({ + f"{Path(sysconfig.get_path(name))}{os.sep}" + for name in ("stdlib", "platstdlib", "purelib", "platlib") +}) + + +class EventLoopBlockingMonitor: + """Report tasks, such as event handlers, that keep the event loop from responding.""" + + def __init__( + self, + loop: asyncio.AbstractEventLoop, + threshold: float, + event_handlers: Mapping[str, RegisteredEventHandler], + ) -> None: + """Create a monitor for an event loop. + + Args: + loop: The event loop to monitor. + threshold: Blocking duration in seconds at which to warn. + event_handlers: The registered handlers, by event name. + """ + self.loop = loop + self.threshold = threshold + self.event_handlers = event_handlers + # The monitor is created on the loop thread, in the app root. + self._loop_thread_id = threading.get_ident() + self._app_root = f"{Path.cwd()}{os.sep}" + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + """Start the single monitor thread for this event processor.""" + self._thread = threading.Thread( + target=self._run, + name="reflex-event-loop-monitor", + daemon=True, + ) + self._thread.start() + + def stop(self) -> None: + """Stop the monitor thread.""" + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=1) + self._thread = None + + def _describe_blocking(self) -> tuple[str, str, str, str] | None: + """Describe the task that runs on the blocked loop. + + Returns: + The subject (event handler or task name), the scope note for event + handlers (or an empty string), the location of the innermost app + frame (or an empty string), and the formatted stack of the task, or + None if no task runs or the task has left the loop. + """ + task = asyncio.current_task(self.loop) + if task is None or (stack := self._task_stack(task)) is None: + return None + name = task.get_name() + if name.startswith(EVENT_TASK_PREFIX): + event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0] + registered = self.event_handlers.get(event_name) + handler_name = ( + event_name if registered is None else registered.handler.fn.__qualname__ + ) + subject = f"Event handler {handler_name}" + scope = ", including state updates such as computed vars" + else: + subject, scope = f"Task {name!r}", "" + app_frame = self._app_frame(stack) + location = ( + "" + if app_frame is None + else f" at {app_frame.filename[len(self._app_root) :]}:" + f"{app_frame.lineno} in {app_frame.name}" + ) + return subject, scope, location, "".join(traceback.format_list(stack)) + + def _app_frame( + self, stack: traceback.StackSummary + ) -> traceback.FrameSummary | None: + """Find the innermost frame of app code, outside the stdlib and packages. + + Args: + stack: The frames, outermost first. + + Returns: + The innermost app frame, or None if no frame is app code. + """ + for frame in reversed(stack): + if frame.filename.startswith( + self._app_root + ) and not frame.filename.startswith(_LIBRARY_PATHS): + return frame + return None + + def _task_stack(self, task: asyncio.Task) -> traceback.StackSummary | None: + """Extract the loop thread's frames from the task entry to the blocking line. + + Args: + task: The task that runs on the blocked loop. + + Returns: + The frames, outermost first, or None if the loop thread no longer + runs the task. + """ + entry = getattr(task.get_coro(), "cr_frame", None) + frame = sys._current_frames().get(self._loop_thread_id) + frames = [] + while frame is not None: + frames.append(frame) + if frame is entry: + return traceback.StackSummary.extract( + (frame, frame.f_lineno) for frame in reversed(frames) + ) + frame = frame.f_back + return None + + def _run(self) -> None: + """Poll loop responsiveness from outside the event-loop thread.""" + # Poll at one tenth of the threshold, between one and 10 milliseconds. + interval = min(max(self.threshold / 10, 0.001), 0.01) + while not self._stop.wait(interval): + responsive = threading.Event() + started = time.perf_counter() + try: + self.loop.call_soon_threadsafe(responsive.set) + except RuntimeError: + return + + # The task is only visible while the loop is blocked, so report it + # then, and report the total duration once the loop responds. + subject = None + while not responsive.wait(interval): + if self._stop.is_set(): + return + elapsed = time.perf_counter() - started + if ( + subject is None + and elapsed + interval >= self.threshold + and (blocking := self._describe_blocking()) is not None + ): + subject, scope, location, stack = blocking + logger.warning( + "%s is blocking the event loop%s (threshold: %.3fs)%s. " + "Move blocking I/O to a thread pool or use a non-blocking " + "async API.\n" + "Blocking stack (most recent call last):\n%s", + subject, + location, + self.threshold, + scope, + stack.rstrip(), + ) + if subject is not None: + logger.warning( + "%s blocked the event loop for %.3fs in total.", + subject, + time.perf_counter() - started, + ) diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index 51f7e1231be..3b76861899d 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -17,8 +17,13 @@ from reflex.app_mixins.middleware import MiddlewareMixin from reflex.istate.manager import StateManager -from reflex_base import otel +from reflex_base import constants, otel +from reflex_base.environment import environment from reflex_base.event.context import EventContext +from reflex_base.event.processor.blocking import ( + EVENT_TASK_PREFIX, + EventLoopBlockingMonitor, +) from reflex_base.event.processor.future import EventFuture from reflex_base.event.processor.timeout import DrainTimeoutManager from reflex_base.registry import RegisteredEventHandler, RegistrationContext @@ -108,6 +113,7 @@ class EventProcessor: Callable[[Exception], EventSpec | list[EventSpec] | None] | None ) = None graceful_shutdown_timeout: float | None = None + event_loop_blocking_threshold: float = 0 _queue: asyncio.Queue[EventQueueEntry] | None = dataclasses.field( default=None, init=False @@ -137,6 +143,9 @@ class EventProcessor: str, collections.deque[tuple[EventQueueEntry, RegisteredEventHandler]], ] = dataclasses.field(default_factory=dict, init=False) + _blocking_monitor: EventLoopBlockingMonitor | None = dataclasses.field( + default=None, init=False, repr=False + ) def configure( self, @@ -243,6 +252,16 @@ async def start(self) -> None: raise RuntimeError(msg) self._attached_root_context_token = EventContext.set(self._root_context) self._queue = asyncio.Queue() + if ( + self.event_loop_blocking_threshold + and environment.REFLEX_ENV_MODE.get() == constants.Env.DEV + ): + self._blocking_monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), + self.event_loop_blocking_threshold, + RegistrationContext.get().event_handlers, + ) + self._blocking_monitor.start() self._ensure_queue_task() async def _stop_tasks(self, timeout: float | None = None) -> None: @@ -339,6 +358,9 @@ async def stop(self, graceful_shutdown_timeout: float | None = None) -> None: if not future.done(): future.cancel() self._futures.clear() + if self._blocking_monitor is not None: + self._blocking_monitor.stop() + self._blocking_monitor = None async def join( self, timeout: float | None = None, queue: asyncio.Queue | None = None @@ -772,7 +794,9 @@ def _create_event_task( coro = self._process_event_queue_entry( entry=entry, registered_handler=registered_handler ) - name = f"reflex_event|{entry.event.name}|{entry.ctx.token}|{entry.ctx.txid}" + name = ( + f"{EVENT_TASK_PREFIX}{entry.event.name}|{entry.ctx.token}|{entry.ctx.txid}" + ) loop = asyncio.get_running_loop() if ( sys.version_info >= (3, 12) diff --git a/pyproject.toml b/pyproject.toml index 8d59db5a773..1eef5b8d253 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "starlette >=1.3.1", "typing_extensions >=4.13.0", "wrapt >=1.17.0,<2.6", - "reflex-base >= 0.10.1", + "reflex-base >= 0.10.2.dev0", "reflex-components-code >= 0.10.1", "reflex-components-core >= 0.10.0", "reflex-components-dataeditor >= 0.10.1", diff --git a/reflex/app.py b/reflex/app.py index 97d7d7472e5..39ce47ec1c9 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -768,7 +768,9 @@ async def _setup_event_processor(self) -> AsyncIterator[None]: event_namespace._token_manager._reset_instance_id() # Create the event processor. self._event_processor = BaseStateEventProcessor( - middleware=self, backend_exception_handler=self.backend_exception_handler + middleware=self, + backend_exception_handler=self.backend_exception_handler, + event_loop_blocking_threshold=get_config().event_loop_blocking_threshold, ) async with self._event_processor.configure( state_manager=self.state_manager, diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py new file mode 100644 index 00000000000..084c42daf16 --- /dev/null +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -0,0 +1,230 @@ +"""Tests for development-time event-loop blocking detection.""" + +import asyncio +import logging +import sys +import threading +import time +from collections.abc import Callable, Coroutine +from typing import Any + +import pytest +from reflex_base.event.processor.blocking import ( + EVENT_TASK_PREFIX, + EventLoopBlockingMonitor, +) +from reflex_base.registry import RegisteredEventHandler + +from reflex.event import EventHandler + + +def _track_queued_probe( + loop: asyncio.AbstractEventLoop, monkeypatch: pytest.MonkeyPatch +) -> threading.Event: + """Return an event set after the monitor queues a responsiveness probe.""" + probe_queued = threading.Event() + call_soon_threadsafe = loop.call_soon_threadsafe + + def track_probe( + callback: Callable[..., Any], *args: Any, **kwargs: Any + ) -> asyncio.Handle: + """Queue the callback and record that a probe was queued. + + Args: + callback: The callback to queue. + *args: Positional arguments for the callback. + **kwargs: Keyword arguments for call_soon_threadsafe. + + Returns: + The handle of the queued callback. + """ + handle = call_soon_threadsafe(callback, *args, **kwargs) + probe_queued.set() + return handle + + monkeypatch.setattr(loop, "call_soon_threadsafe", track_probe) + return probe_queued + + +async def _run_monitored( + fn: Callable[[], Coroutine[Any, Any, None]], + task_name: str, + event_handlers: dict[str, RegisteredEventHandler] | None = None, +) -> None: + """Run fn in a task with the given name while a monitor watches the loop.""" + monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), threshold=0.04, event_handlers=event_handlers or {} + ) + monitor.start() + try: + await asyncio.create_task(fn(), name=task_name) + await asyncio.sleep(0) + finally: + monitor.stop() + + +def _blocking( + probe_queued: threading.Event, duration: float +) -> Callable[[], Coroutine[Any, Any, None]]: + """Create a coroutine function that blocks the loop after a probe is queued. + + Returns: + The blocking coroutine function. + """ + + async def blocking_handler(): # noqa: RUF029 + """Block the loop for the duration after a probe is queued.""" + assert probe_queued.wait(timeout=1) + time.sleep(duration) # noqa: ASYNC251 + + return blocking_handler + + +@pytest.fixture +def caplog_blocking(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture: + """Capture warnings of the blocking monitor. + + Returns: + The log capture fixture. + """ + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + return caplog + + +async def test_blocking_event_is_reported_with_handler_name( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Name the registered handler of the blocking event task.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + handler = _blocking(probe_queued, 0.12) + registered = RegisteredEventHandler(handler=EventHandler(fn=handler), states=()) + + await _run_monitored( + handler, + f"{EVENT_TASK_PREFIX}state.blocking|token|txid", + {"state.blocking": registered}, + ) + + assert "blocking_handler" in caplog_blocking.text + assert "blocked the event loop for 0.1" in caplog_blocking.text + assert "time.sleep(duration) # noqa: ASYNC251" in caplog_blocking.text + assert "_run_monitored" not in caplog_blocking.text + assert "thread pool or use a non-blocking async API" in caplog_blocking.text + + +async def test_unregistered_event_is_reported_with_event_name( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Fall back to the event name when no handler is registered for it.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + + await _run_monitored( + _blocking(probe_queued, 0.12), f"{EVENT_TASK_PREFIX}state.unknown|token|txid" + ) + + assert "Event handler state.unknown blocked" in caplog_blocking.text + + +async def test_blocking_just_over_threshold_is_detected( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Account for the time between responsiveness probes.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + + await _run_monitored( + _blocking(probe_queued, 0.055), f"{EVENT_TASK_PREFIX}state.blocking|token|txid" + ) + + assert "state.blocking" in caplog_blocking.text + + +async def test_blocking_outside_event_task_is_reported_with_task_name( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Report blocking in a task that processes no event by the task name.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + + await _run_monitored(_blocking(probe_queued, 0.12), "other_task") + + assert "Task 'other_task' is blocking the event loop" in caplog_blocking.text + assert "Task 'other_task' blocked the event loop for 0.1" in caplog_blocking.text + assert "time.sleep(duration) # noqa: ASYNC251" in caplog_blocking.text + assert "Event handler" not in caplog_blocking.text + assert "computed vars" not in caplog_blocking.text + + +async def test_awaiting_event_is_not_reported( + caplog_blocking: pytest.LogCaptureFixture, +): + """Do not report handlers that yield control while doing async work.""" + + async def yielding_handler(): + """Yield to the loop while waiting.""" + await asyncio.sleep(0.1) + + await _run_monitored( + yielding_handler, f"{EVENT_TASK_PREFIX}state.yielding|token|txid" + ) + + assert "the event loop" not in caplog_blocking.text + + +async def test_hung_event_is_reported_while_still_blocked( + caplog_blocking: pytest.LogCaptureFixture, +): + """Warn when the threshold is reached, before the loop responds again.""" + warned = threading.Event() + + class _SetOnWarning(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + """Record that the monitor logged a warning. + + Args: + record: The log record. + """ + warned.set() + + handler = _SetOnWarning(level=logging.WARNING) + blocking_logger = logging.getLogger("reflex_base.event.processor.blocking") + blocking_logger.addHandler(handler) + + async def hung_handler(): # noqa: RUF029 + """Hold the loop until the monitor warns, like a handler that never returns.""" + assert warned.wait(timeout=1) + + try: + await _run_monitored(hung_handler, f"{EVENT_TASK_PREFIX}state.hung|token|txid") + finally: + blocking_logger.removeHandler(handler) + + assert "state.hung is blocking the event loop" in caplog_blocking.text + assert "warned.wait(timeout=1)" in caplog_blocking.text + + +async def test_stack_without_task_entry_is_not_reported( + monkeypatch: pytest.MonkeyPatch, +): + """Do not describe a frame snapshot taken after the task left the loop.""" + monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), threshold=0.04, event_handlers={} + ) + unrelated = sys._getframe() + monkeypatch.setattr( + sys, + "_current_frames", + lambda: {monitor._loop_thread_id: unrelated}, + ) + + async def handler(): # noqa: RUF029 + """Describe the blocking from inside the event task. + + Returns: + The description of the blocking. + """ + return monitor._describe_blocking() + + description = await asyncio.create_task( + handler(), name=f"{EVENT_TASK_PREFIX}state.moved|token|txid" + ) + + assert description is None diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index e7a389b4392..b009404fe8d 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -4,15 +4,19 @@ import contextlib import dataclasses import logging +import re import sys +import threading +import time from collections.abc import AsyncIterator, Mapping +from pathlib import Path from typing import Any from unittest.mock import Mock import pytest from opentelemetry.trace import SpanKind from pytest_mock import MockerFixture -from reflex_base import otel +from reflex_base import constants, otel from reflex_base.event.context import EventContext from reflex_base.event.processor.event_processor import ( EventProcessor, @@ -42,6 +46,16 @@ async def _slow_handler(delay: float = 0.5): await asyncio.sleep(delay) +def _blocking_sync_handler(): + """Block the event loop briefly for the detector integration test.""" + time.sleep(0.12) + + +def _blocking_stdlib_handler(): + """Block the event loop inside a stdlib call for the detector test.""" + threading.Event().wait(0.12) + + async def _error_handler(): # noqa: RUF029 """A handler that always raises.""" raise RuntimeError("boom") # noqa: EM101 @@ -364,6 +378,8 @@ async def _counting_superseding_handler(tick: int = 0, limit: int = 3): noop_event = EventHandler(fn=_noop_handler) slow_event = EventHandler(fn=_slow_handler) +blocking_sync_event = EventHandler(fn=_blocking_sync_handler) +blocking_stdlib_event = EventHandler(fn=_blocking_stdlib_handler) error_event = EventHandler(fn=_error_handler) recovery_event = EventHandler(fn=_recovery_handler) logging_event = EventHandler(fn=_logging_handler) @@ -404,6 +420,8 @@ def _register_handlers(forked_registration_context: RegistrationContext): for handler in ( noop_event, slow_event, + blocking_sync_event, + blocking_stdlib_event, error_event, recovery_event, logging_event, @@ -517,6 +535,33 @@ async def test_start_twice(processor: EventProcessor): await processor.stop() +async def test_blocking_monitor_starts_only_in_development( + processor: EventProcessor, monkeypatch: pytest.MonkeyPatch +): + """Enable the loop monitor in dev and leave it off in production.""" + processor.event_loop_blocking_threshold = 0.05 + processor.configure() + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.PROD.value) + await processor.start() + assert processor._blocking_monitor is None + await processor.stop() + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + dev_processor = EventProcessor(event_loop_blocking_threshold=0.05) + dev_processor.configure() + await dev_processor.start() + assert dev_processor._blocking_monitor is not None + await dev_processor.stop() + assert dev_processor._blocking_monitor is None + + disabled_processor = EventProcessor(event_loop_blocking_threshold=0) + disabled_processor.configure() + await disabled_processor.start() + assert disabled_processor._blocking_monitor is None + await disabled_processor.stop() + + async def test_stop_idempotent(processor: EventProcessor): """Stopping an already-stopped processor does not error. @@ -613,6 +658,78 @@ async def test_events_are_processed( assert _CALL_LOG == [{"value": "hello"}] +async def test_processor_reports_blocking_sync_handler( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Report blocking handlers executed through the event queue.""" + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(blocking_sync_event())[0]) + + assert "_blocking_sync_handler" in caplog.text + assert "blocked the event loop" in caplog.text + duration = re.search(r"blocked the event loop for (\d+\.\d+)s", caplog.text) + assert duration is not None + assert float(duration.group(1)) >= 0.1 + # The stack of the event task ends at the blocking line, without loop frames. + assert "Blocking stack (most recent call last):" in caplog.text + assert "time.sleep(0.12)" in caplog.text + assert "base_events.py" not in caplog.text + + +async def test_processor_reports_innermost_app_frame( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Name the app line that called the blocking library code.""" + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(blocking_stdlib_event())[0]) + + path = Path( + "tests", "units", "reflex_base", "event", "processor", "test_event_processor.py" + ) + assert f"is blocking the event loop at {path}:" in caplog.text + assert "in _blocking_stdlib_handler (threshold" in caplog.text + + +async def test_processor_reports_blocking_after_handler_returns( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Attribute blocking in post-handler processing (e.g. computed vars) to the event.""" + execute_event = EventProcessor._execute_event + + async def _execute_event(self: EventProcessor, **kwargs): + await execute_event(self, **kwargs) + time.sleep(0.12) # noqa: ASYNC251 + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + monkeypatch.setattr(EventProcessor, "_execute_event", _execute_event) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(noop_event())[0]) + + assert "_noop_handler" in caplog.text + assert "blocked the event loop" in caplog.text + assert "time.sleep(0.12) # noqa: ASYNC251" in caplog.text + + async def test_enqueue_returns_future( mock_event_processor: EventProcessor, token: str, diff --git a/tests/units/test_app.py b/tests/units/test_app.py index ae575628094..20ce7ff3d9f 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -3520,6 +3520,19 @@ def test_app_state_determination(): assert a2._state is None +async def test_setup_event_processor_passes_blocking_threshold( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture +): + """Forward the app config threshold to its event processor.""" + app = App() + config = rx.Config(app_name="blockingthreshold", event_loop_blocking_threshold=0.35) + mocker.patch("reflex.app.get_config", return_value=config) + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.PROD.value) + + async with app._setup_event_processor(): + assert app.event_processor.event_loop_blocking_threshold == pytest.approx(0.35) + + def test_raise_on_state(): """Test that the state is set.""" # state kwargs is deprecated, we just make sure the app is created anyway. diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 4ec4bc3d68c..2d94059226f 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -72,6 +72,41 @@ def test_trusted_proxy_hops_defaults_to_zero(base_config_values): assert rx.Config(**base_config_values).trusted_proxy_hops == 0 +def test_event_loop_blocking_threshold_defaults_to_development_threshold( + base_config_values, +): + """Enable useful blocking diagnostics by default in development.""" + config = rx.Config(**base_config_values) + assert config.event_loop_blocking_threshold == pytest.approx(0.2) + + +@pytest.mark.parametrize("threshold", [-1, float("inf"), float("nan"), True, "1", None]) +def test_event_loop_blocking_threshold_must_be_non_negative_finite( + base_config_values, threshold +): + """Reject invalid event-loop blocking thresholds.""" + with pytest.raises( + ConfigError, + match="event_loop_blocking_threshold must be a non-negative finite number", + ): + rx.Config(**base_config_values, event_loop_blocking_threshold=threshold) + + +def test_event_loop_blocking_detection_can_be_disabled(base_config_values): + """Allow apps to disable event-loop blocking detection with 0.""" + config = rx.Config(**base_config_values, event_loop_blocking_threshold=0) + assert config.event_loop_blocking_threshold == 0 + + +def test_event_loop_blocking_detection_can_be_disabled_from_env( + base_config_values, monkeypatch: pytest.MonkeyPatch +): + """Allow the environment to disable event-loop blocking detection.""" + monkeypatch.setenv("REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD", "0") + config = rx.Config(**base_config_values) + assert config.update_from_env()["event_loop_blocking_threshold"] == 0 + + @pytest.mark.parametrize("trusted_proxy_hops", [-1, True, False, 1.5, "2", None]) def test_trusted_proxy_hops_must_be_non_negative( base_config_values, trusted_proxy_hops @@ -114,6 +149,7 @@ def test_react_compiler_default(base_config_values): ("REFLEX_REACT_COMPILER", False), ("REFLEX_REACT_COMPILER", True), ("REFLEX_DEFAULT_COLOR_MODE", "dark"), + ("REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD", 0.35), ], ) def test_update_from_env(