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..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 @@ -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] @@ -970,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: @@ -978,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 @@ -1014,6 +1027,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..8f4b1cd34d0 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,109 @@ 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_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,