From 58ac79e276536cbc3fcaeb757be4f9db7407f3cf Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 29 Oct 2026 22:07:54 +0500 Subject: [PATCH] fix(event): send a failed handler's writes after its last yield A foreground handler that is cancelled or raises could leave changes it made after its last yield unsent. Memory and disk carried them in the next handler's delta, but a state manager that loads a fresh copy, as redis does, dropped the dirty marks, so the frontend never received them until the var changed again or the page reloaded. Flush the delta under the state lock before the exception propagates. A flush failure is logged so the handler's own exception still propagates. --- .../+flush-failed-handler-delta.bugfix.md | 1 + .../event/processor/base_state_processor.py | 36 ++++- .../processor/test_base_state_processor.py | 142 ++++++++++++++++++ 3 files changed, 173 insertions(+), 6 deletions(-) create mode 100644 packages/reflex-base/news/+flush-failed-handler-delta.bugfix.md diff --git a/packages/reflex-base/news/+flush-failed-handler-delta.bugfix.md b/packages/reflex-base/news/+flush-failed-handler-delta.bugfix.md new file mode 100644 index 00000000000..e6591b88972 --- /dev/null +++ b/packages/reflex-base/news/+flush-failed-handler-delta.bugfix.md @@ -0,0 +1 @@ +Send the state changes a foreground event handler makes after its last `yield` when it is cancelled or raises, so the frontend receives them right away. With Redis, they were otherwise not sent until the same var changed again or the page reloaded. diff --git a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py index 7d13b83e437..aec40ceb1f3 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import dataclasses import functools import inspect @@ -281,6 +282,23 @@ async def chain_updates( await _route_events(ctx, fixed_events) +async def _flush_failed_handler(root_state: BaseState, handler_name: str) -> None: + """Send the writes a cancelled or failing handler made after its last yield. + + The next handler's delta does not carry them when it loads a fresh copy of + the state, as redis does. A flush failure is logged, so the handler's own + exception stays the one that propagates. + + Args: + root_state: The locked root state of the handler. + handler_name: The name of the handler, used for error messages. + """ + try: + await chain_updates(None, root_state=root_state, handler_name=handler_name) + except Exception: + logger.exception(f"Error flushing delta after handler {handler_name} failed:") + + def ensure_locked( state: BaseState | StateProxy, root_state: BaseState | None ) -> BaseState | None: @@ -597,12 +615,18 @@ async def _execute_event( # Process non-background events while holding the lock. if not registered_handler.handler.is_background: - await process_event( - handler=registered_handler.handler, - payload=event.payload, - state=substate, - root_state=root_state, - ) + try: + await process_event( + handler=registered_handler.handler, + payload=event.payload, + state=substate, + root_state=root_state, + ) + except (Exception, asyncio.CancelledError): + await _flush_failed_handler( + root_state, registered_handler.handler.fn.__qualname__ + ) + raise return # Otherwise drop the state lock and start processing the background task # with a proxy state. No root_state: the lock is no longer held, and diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index ebbbc91f23d..ec593420a45 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -1418,6 +1418,148 @@ def ping(self): assert (await root.get_state(State)).is_hydrated is True +@pytest.mark.parametrize("outcome", ["cancelled", "raised"]) +@pytest.mark.parametrize( + "processor_state_manager", ["in_process", "disk", "redis"], indirect=True +) +async def test_failed_handler_emits_writes_after_last_yield( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + processor_state_manager: StateManager, + emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], + token: str, + outcome: str, +): + """A cancelled or failing handler sends the writes it made after its last yield. + + Args: + wired_app: The App wired to the processor's state manager. + real_base_state_processor: The unmocked event processor. + processor_state_manager: The state manager backing the processor. + emitted_deltas: The deltas sent by the processor. + token: The client token. + outcome: Whether the handler is cancelled or raises. + """ + started = asyncio.Event() + + class CleanupState(State): + """Write a var only after the handler's last yield.""" + + cleanup: str = "" + + @event + async def run(self): + """Yield, then write in cleanup when cancelled or failing. + + Yields: + The state update before waiting. + + Raises: + RuntimeError: When the handler is set to fail. + """ + yield + if outcome == "raised": + self.cleanup = "raised" + msg = "handler failed" + raise RuntimeError(msg) + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.cleanup = "cancelled" + raise + + handled: list[Exception] = [] + real_base_state_processor.backend_exception_handler = handled.append + wired_app.add_page(lambda: rx.text("cleanup"), route="/") + async with real_base_state_processor as processor: + future = await processor.enqueue( + token, _client_event(CleanupState.run(), _view("/")) + ) + if outcome == "cancelled": + await asyncio.wait_for(started.wait(), timeout=5) + future.cancel() + with contextlib.suppress(BaseException): + await asyncio.wait_for(future.wait_all(), timeout=5) + + field = "cleanup" + FIELD_MARKER + sent = [ + fields[field] + for _, delta in emitted_deltas + if field in (fields := delta.get(CleanupState.get_full_name(), {})) + ] + assert sent[-1] == outcome + # The handler's own exception is reported, not one from the flush. + assert [str(ex) for ex in handled] == ( + ["handler failed"] if outcome == "raised" else [] + ) + + +async def test_failed_handler_flush_error_keeps_handler_exception( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + token: str, + caplog: pytest.LogCaptureFixture, +): + """A failing flush is logged and the handler's own exception is reported. + + Args: + wired_app: The App wired to the processor's state manager. + real_base_state_processor: The unmocked event processor. + token: The client token. + caplog: The pytest log capture fixture. + """ + + class FlushErrorState(State): + """Fail the delta computation once the handler wrote its cleanup.""" + + cleanup: str = "" + + @rx.var + def checked(self) -> str: + """Raise once cleanup is set. + + Returns: + The cleanup value. + + Raises: + ValueError: When cleanup is set. + """ + if self.cleanup: + msg = "delta failed" + raise ValueError(msg) + return self.cleanup + + @event + def run(self): + """Write cleanup, then fail. + + Raises: + RuntimeError: Always. + """ + self.cleanup = "raised" + msg = "handler failed" + raise RuntimeError(msg) + + handled: list[Exception] = [] + real_base_state_processor.backend_exception_handler = handled.append + wired_app.add_page(lambda: rx.text("flush"), route="/") + with caplog.at_level(logging.ERROR): + async with real_base_state_processor as processor: + future = await processor.enqueue( + token, _client_event(FlushErrorState.run(), _view("/")) + ) + with contextlib.suppress(RuntimeError): + await asyncio.wait_for(future.wait_all(), timeout=5) + + assert [str(ex) for ex in handled] == ["handler failed"] + assert any( + "Error flushing delta after handler" in record.getMessage() + and "delta failed" in (record.exc_text or "") + for record in caplog.records + ) + + async def test_rehydrate_after_expiry_does_not_reload_the_previous_route( wired_app: App, real_base_state_processor: BaseStateEventProcessor,