From 20cc4b64ae6abd7a9240bf8b757ed059d7f853f1 Mon Sep 17 00:00:00 2001 From: Farhan Date: Sun, 15 Nov 2026 09:14:22 +0000 Subject: [PATCH 1/2] Fix awaiting an event handler from a background task An exception raised by a handler awaited from a background task (no state lock held, so the event is enqueued) was delivered to the awaiting task and also reported to the backend exception handler, showing the error toast even when the task caught it. The enqueued event's future is now marked as awaited, and its exception is left to the caller: if the task does not catch it, the task's own failure reports it once. A supersedes=True handler run that had returned, but whose task done callback had not yet resolved its future, was cancelled by a newer invocation enqueued in that window, so the awaiting task got CancelledError for a run that completed. The in-memory state manager hits the window because a task waiting for the state lock resumes without suspending before it enqueues. Superseding such a run now only cancels the events it chained. Closes #7839 Claude-Session: https://claude.ai/code/session_01LwJUCU8RtEco15mQvkxdQb --- ...+awaited-handler-from-background.bugfix.md | 1 + .../event/processor/base_state_processor.py | 3 + .../event/processor/event_processor.py | 14 +- .../src/reflex_base/event/processor/future.py | 5 + .../tests_playwright/test_background_task.py | 45 ++++++ .../processor/test_base_state_processor.py | 128 ++++++++++++++++++ .../event/processor/test_event_processor.py | 60 ++++++++ 7 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 packages/reflex-base/news/+awaited-handler-from-background.bugfix.md diff --git a/packages/reflex-base/news/+awaited-handler-from-background.bugfix.md b/packages/reflex-base/news/+awaited-handler-from-background.bugfix.md new file mode 100644 index 00000000000..fa289ed085d --- /dev/null +++ b/packages/reflex-base/news/+awaited-handler-from-background.bugfix.md @@ -0,0 +1 @@ +Awaiting an event handler from a background task now behaves like the inline call: an exception the task catches is no longer also reported to the backend exception handler (no error toast), and a `supersedes=True` handler run that finished before a newer one superseded it returns its result instead of raising `CancelledError`. 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 057e5a4d452..dc22c29a700 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 @@ -616,6 +616,9 @@ async def execute_event_inline(spec: EventSpec) -> Any: root_state = ctx.locked_root if root_state is None: (future,) = await ctx.enqueue(event) + # An exception the event raises is delivered here, like the inline + # call's, so it is this caller's to catch or to fail with. + future.awaited = True return await future if handler.is_background: msg = ( 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 a3339086e2c..01f0be94aef 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 @@ -704,7 +704,15 @@ def _cancel_older_chains(self, key: tuple[str, str], root_gen: int) -> None: f"{key[1]}, superseded by a newer invocation." ) for previous in slot.values(): - previous.cancel() + task = self._tasks.get(previous.txid) + if task is not None and task.done() and not previous.done(): + # The run finished, and its done callback has yet to resolve + # the future with its outcome: only the events it chained + # are left to supersede. + for child in previous.children: + child.cancel() + else: + previous.cancel() slot.clear() def _on_future_done(self, future: EventFuture) -> None: # type: ignore[override] @@ -1014,6 +1022,10 @@ def _resolve_future( with contextlib.suppress(BaseException): # Trigger the future to avoid warnings if the caller didn't wait. future.result() + if future.awaited: + # The caller awaiting the event received the exception: + # reporting is its failure path's, if it does not catch it. + return False telemetry.send_error(ex, context="backend") if ( not task.get_name().startswith("reflex_backend_exception_handler|") diff --git a/packages/reflex-base/src/reflex_base/event/processor/future.py b/packages/reflex-base/src/reflex_base/event/processor/future.py index a05730b0fff..423cb786a45 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/future.py +++ b/packages/reflex-base/src/reflex_base/event/processor/future.py @@ -47,6 +47,11 @@ class EventFuture(asyncio.Future): default=frozenset(), repr=False ) + # Whether a caller awaits this future as a function call (awaiting an event + # handler with no state lock held), so an exception the event raises is + # the caller's to handle rather than the backend exception handler's. + awaited: bool = dataclasses.field(default=False, repr=False) + def __post_init__(self) -> None: """Call Future.__init__ for the EventFuture.""" super(EventFuture, self).__init__(loop=self.loop) diff --git a/tests/integration/tests_playwright/test_background_task.py b/tests/integration/tests_playwright/test_background_task.py index b8e0dc8b475..68756a20113 100644 --- a/tests/integration/tests_playwright/test_background_task.py +++ b/tests/integration/tests_playwright/test_background_task.py @@ -125,6 +125,20 @@ async def await_handlers(self): assert self.counter == 3 await State.increment_by(await State.pick_amount(2)) + @rx.event(background=True) + async def await_failing_handler(self): + try: + await State.failing_increment() + except ValueError: + async with self: + self.counter += 10 + + @rx.event + def failing_increment(self): + self.counter += 1 + msg = "caught by the awaiting background task" + raise ValueError(msg) + @rx.event def increment_by(self, amount: int): self.counter += amount @@ -236,6 +250,11 @@ def index() -> rx.Component: on_click=State.await_handlers, id="await-handlers", ), + rx.button( + "Await Failing Handler", + on_click=State.await_failing_handler, + id="await-failing-handler", + ), rx.button("Reset", on_click=State.reset_counter, id="reset"), ) @@ -457,6 +476,32 @@ def test_await_handlers_from_background_task( expect(counter).to_have_text("7", timeout=5_000) +def test_await_failing_handler_from_background_task( + background_task: AppHarness, + background_page: Page, + token: str, +): + """An awaited handler's exception that the background task catches shows no toast. + + Args: + background_task: harness for BackgroundTask app. + background_page: Playwright page. + token: The token for the connected client. + """ + assert background_task.app_instance is not None + + counter = background_page.locator("#counter") + expect(counter).to_have_text("0") + + background_page.locator("#await-failing-handler").click() + # The failed handler's write is kept, then the task handles the exception. + expect(counter).to_have_text("11", timeout=5_000) + # A sequential event after it, so an error toast would have arrived by now. + background_page.locator("#increment").click() + expect(counter).to_have_text("12") + expect(background_page.locator("[data-sonner-toast]")).to_have_count(0) + + @pytest.mark.parametrize( "button_id", [ 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 bf2c7ac8c63..6241eed0b66 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 @@ -5582,3 +5582,131 @@ def ping(self): BaseStateToken(ident=token, cls=State) ) assert (await state.get_state(CancelInsideState)).log == ["ping"] + + +@pytest.mark.parametrize( + ("catches", "expected_results", "expected_reported"), + [ + (True, ["ValueError('boom')"], []), + (False, [], ["ValueError('boom')"]), + ], +) +async def test_exception_of_handler_awaited_from_background_is_the_callers( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + token: str, + catches: bool, + expected_results: list[str], + expected_reported: list[str], +): + """An awaited handler's exception reaches the backend handler only if uncaught. + + The background task awaits with no lock held, so the handler is enqueued. + Its exception is raised from the await like the inline call's: caught, it + is not reported; uncaught, the background task's failure reports it once. + + Args: + wired_app: The App wired to the processor's state manager. + real_base_state_processor: The unmocked BaseStateEventProcessor. + token: The client token. + catches: Whether the background task catches the exception. + expected_results: The exceptions the background task caught. + expected_reported: The exceptions the backend exception handler got. + """ + results: list[str] = [] + reported: list[str] = [] + + class BgBoomState(State): + total: int = 0 + + @event + def boom(self): + self.total += 1000 + msg = "boom" + raise ValueError(msg) + + @event(background=True) + async def run(self, catches: bool): + try: + await BgBoomState.boom() + except ValueError as ex: + if not catches: + raise + results.append(repr(ex)) + + real_base_state_processor.backend_exception_handler = lambda ex: reported.append( + repr(ex) + ) + async with real_base_state_processor as processor: + future = await processor.enqueue( + token, Event.from_event_type(BgBoomState.run(catches))[0] + ) + with contextlib.suppress(ValueError): + await future.wait_all() + + assert results == expected_results + assert reported == expected_reported + async with _read_back(real_base_state_processor, token) as root: + assert (await root.get_state(BgBoomState)).total == 1000 + + +@pytest.mark.parametrize( + "processor_state_manager", ["in_process", "redis"], indirect=True +) +async def test_superseding_handler_awaited_from_background_returns_finished_result( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + token: str, +): + """A background task gets the result of a superseding handler run that finished. + + The second task waits for the state lock the first search holds, and + supersedes it as soon as that search has returned. With the in-memory + manager that is before the first search's future was resolved. + + Args: + wired_app: The App wired to the processor's state manager. + real_base_state_processor: The unmocked BaseStateEventProcessor. + token: The client token. + """ + results: list[str] = [] + started = asyncio.Event() + release = asyncio.Event() + + class BgSearchState(State): + query: str = "" + + @event(supersedes=True) + async def search(self, q: str) -> str: + if q == "first": + started.set() + await release.wait() + self.query = q + return q.upper() + + @event(background=True) + async def run(self, q: str): + try: + result = await BgSearchState.search(q) + except asyncio.CancelledError: + result = "cancelled" + results.append(f"{q} -> {result}") + + async with real_base_state_processor as processor: + first = await processor.enqueue( + token, Event.from_event_type(BgSearchState.run("first"))[0] + ) + await asyncio.wait_for(started.wait(), timeout=5) + second = await processor.enqueue( + token, Event.from_event_type(BgSearchState.run("second"))[0] + ) + # Let the second task start and queue for the lock the search holds. + await asyncio.sleep(0.1) + release.set() + await asyncio.wait_for( + asyncio.gather(first.wait_all(), second.wait_all()), timeout=5 + ) + + assert results == ["first -> FIRST", "second -> SECOND"] + async with _read_back(real_base_state_processor, token) as root: + assert (await root.get_state(BgSearchState)).query == "second" 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 7e5292a6036..7950010017c 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -1508,6 +1508,66 @@ async def test_superseding_event_cancels_previous_chain( assert _CALL_LOG == [{"value": "stale_cancelled"}, {"value": "fresh"}] +async def test_superseding_event_keeps_the_result_of_a_finished_run( + processor: EventProcessor, + token: str, +): + """A run that finished before it was superseded resolves with its result. + + The newer invocation is enqueued after the older handler returned and + before its task's done callback resolved the future. Only the events the + finished run chained are cancelled. + + Args: + processor: The event processor fixture. + token: The client token. + """ + release = asyncio.Event() + returning = asyncio.Event() + + async def _search(value: str = "default", chain: bool = False) -> str: + if chain: + await EventContext.get().enqueue( + Event.from_event_type(logging_event(f"{value}_chained"))[0] + ) + await release.wait() + # Wakes the superseding enqueue ahead of this task's done callbacks. + returning.set() + return value.upper() + + _search._reflex_supersedes = True # type: ignore[attr-defined] + search_event = EventHandler(fn=_search) + RegistrationContext.register_event_handler(search_event) + + processor.configure() + async with processor as ep: + first = await ep.enqueue( + token, Event.from_event_type(search_event("first", True))[0] + ) + + async def supersede() -> EventFuture: + await returning.wait() + assert ep._tasks[first.txid].done() + assert not first.done() + return await ep.enqueue( + token, Event.from_event_type(search_event("second"))[0] + ) + + superseding = asyncio.create_task(supersede()) + # Park the superseding task on the gate before the handler sets it. + await asyncio.sleep(0) + release.set() + second = await asyncio.wait_for(superseding, timeout=1) + assert await asyncio.wait_for(first, timeout=1) == "FIRST" + assert await asyncio.wait_for(second.wait_all(), timeout=1) == "SECOND" + (chained,) = first.children + assert chained.cancelled() + await _drain_superseded(ep) + assert ep._superseded == {} + + assert _CALL_LOG == [] + + async def test_superseding_event_logs_debug_on_cancel( processor: EventProcessor, token: str, From 231713f8e4c0d81738754463d0a814210020cd66 Mon Sep 17 00:00:00 2001 From: Farhan Date: Sun, 15 Nov 2026 09:23:20 +0000 Subject: [PATCH 2/2] Resolve a finished event's future before starting the next queued one The next queued handler for a token starts eagerly on Python 3.12+, and could supersede the run that just finished while its future was still unresolved and its task already forgotten, cancelling a completed run. Claude-Session: https://claude.ai/code/session_01LwJUCU8RtEco15mQvkxdQb --- .../event/processor/event_processor.py | 9 +++- .../event/processor/test_event_processor.py | 43 +++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) 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 01f0be94aef..915c0e05bcd 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 @@ -978,6 +978,12 @@ def _finish_task(self, task: asyncio.Task): else: task_ctx = task.get_context().run(EventContext.get) self._tasks.pop(task_ctx.txid, None) + future = self._futures.get(task_ctx.txid) + # Before the next event starts: a handler started eagerly below can + # supersede this one, which must not cancel a future still unresolved. + handling_exception = task.done() and self._resolve_future( + task, task_ctx, future + ) # Chain the next sequential event for this token if applicable. token_queue = self._token_queues.get(task_ctx.token) if token_queue and token_queue[0][0].ctx.txid == task_ctx.txid: @@ -986,8 +992,7 @@ def _finish_task(self, task: asyncio.Task): self._dispatch_next_for_token(task_ctx.token) else: del self._token_queues[task_ctx.token] - future = self._futures.get(task_ctx.txid) - if task.done() and self._resolve_future(task, task_ctx, future): + if handling_exception: return if future is not None: # The task is gone; clean up now in case the future resolved 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 7950010017c..8f4b1cd34d0 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -1568,6 +1568,49 @@ async def supersede() -> EventFuture: assert _CALL_LOG == [] +async def test_queued_handler_superseding_a_finished_run_keeps_its_result( + processor: EventProcessor, + token: str, +): + """The next queued handler cannot cancel the finished run it follows. + + The handler queued behind a superseding run starts as soon as that run's + task finishes (eagerly on Python 3.12+), and supersedes it before + suspending. The finished run still resolves with its result. + + Args: + processor: The event processor fixture. + token: The client token. + """ + release = asyncio.Event() + + async def _search(value: str = "default") -> str: + if value == "first": + await release.wait() + return value.upper() + + async def _search_again(): + await EventContext.get().enqueue( + Event.from_event_type(search_event("second"))[0] + ) + + _search._reflex_supersedes = True # type: ignore[attr-defined] + search_event = EventHandler(fn=_search) + search_again_event = EventHandler(fn=_search_again) + RegistrationContext.register_event_handler(search_event) + RegistrationContext.register_event_handler(search_again_event) + + processor.configure() + async with processor as ep: + first = await ep.enqueue(token, Event.from_event_type(search_event("first"))[0]) + again = await ep.enqueue(token, Event.from_event_type(search_again_event())[0]) + release.set() + assert await asyncio.wait_for(first, timeout=1) == "FIRST" + await asyncio.wait_for(again.wait_all(), timeout=1) + (second,) = again.children + assert second.result() == "SECOND" + + async def test_superseding_event_logs_debug_on_cancel( processor: EventProcessor, token: str,