Fix awaiting an event handler from a background task: caught exceptions and finished superseding runs

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`.

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 = (

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|")

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)

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",
     [

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"

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,