fix: cancel stale on_load chains via handler-declared supersession

A page's on_load chain kept running (and blocked the per-token queue)
after the user navigated away (reflex-dev#6593). Instead of teaching the generic
EventProcessor about page loads, add a handler-declared supersedes
marker (mirroring the background-task marker): enqueuing a superseding
handler cancels the previous unfinished chain for the same handler and
client token. on_load_internal is the first handler to opt in.

Also tighten the cancellation lifecycle so a cancelled chain cannot
leak work: futures stay tracked while their task unwinds (late-chained
events find their cancelled parent and are dropped), and the dequeue
paths treat an already-cleaned future as cancelled instead of
executing it.

Covered by processor unit tests, a state-level end-to-end test, and a
Playwright integration test that navigates away mid-on_load and
asserts the stale handler is cancelled while the new page loads
promptly.

Author: FarhanAliRaza

news/6593.bugfix.md

@@ -0,0 +1 @@
+Cancel the previous unfinished `on_load` event chain when a newer page navigation arrives for the same client, instead of letting stale page-load work block and outlive the navigation.

packages/reflex-base/src/reflex_base/event/__init__.py

@@ -185,6 +185,7 @@ def from_event_type(
 )
 
 BACKGROUND_TASK_MARKER = "_reflex_background_task"
+SUPERSEDES_MARKER = "_reflex_supersedes"
 EVENT_ACTIONS_MARKER = "_rx_event_actions"
 UPLOAD_FILES_CLIENT_HANDLER = "uploadFiles"
 
@@ -442,6 +443,18 @@ def is_background(self) -> bool:
         """
         return getattr(self.fn, BACKGROUND_TASK_MARKER, False)
 
+    @property
+    def supersedes(self) -> bool:
+        """Whether a newer invocation supersedes an older unfinished one.
+
+        When True, enqueuing this handler cancels the previous unfinished
+        event chain rooted at the same handler for the same client token.
+
+        Returns:
+            True if the event handler is marked as superseding.
+        """
+        return getattr(self.fn, SUPERSEDES_MARKER, False)
+
     def __call__(self, *args: Any, **kwargs: Any) -> "EventSpec":
         """Pass arguments to the handler to get an event spec.
 
@@ -2778,6 +2791,7 @@ class EventNamespace:
 
     # Constants
     BACKGROUND_TASK_MARKER = BACKGROUND_TASK_MARKER
+    SUPERSEDES_MARKER = SUPERSEDES_MARKER
     EVENT_ACTIONS_MARKER = EVENT_ACTIONS_MARKER
     _EVENT_FIELDS = _EVENT_FIELDS
     FORM_DATA = FORM_DATA

packages/reflex-base/src/reflex_base/event/processor/event_processor.py

@@ -120,6 +120,11 @@ class EventProcessor:
     _futures: dict[str, EventFuture] = dataclasses.field(
         default_factory=dict, init=False
     )
+    # Latest-wins tracking for superseding handlers: (event name, token) -> the
+    # currently active chain root future.
+    _superseded: dict[tuple[str, str], EventFuture] = dataclasses.field(
+        default_factory=dict, init=False
+    )
     _token_queues: dict[
         str,
         collections.deque[tuple[EventQueueEntry, RegisteredEventHandler]],
@@ -311,6 +316,7 @@ async def stop(self, graceful_shutdown_timeout: float | None = None) -> None:
             self._queue_task = None
         # Discard any pending per-token queue entries.
         self._token_queues.clear()
+        self._superseded.clear()
         # Cancel any remaining unresolved futures.
         for future in self._futures.values():
             if not future.done():
@@ -372,6 +378,8 @@ async def enqueue(
 
         Returns:
             An EventFuture that resolves to the result of the associated task.
+            If the event was chained from an already-cancelled chain, the
+            returned future is already cancelled and the event is dropped.
         """
         if ev_ctx is None:
             try:
@@ -395,7 +403,13 @@ async def enqueue(
         tracked.add_done_callback(self._on_future_done)
         # If this context has a parent, register as a child of the parent's future.
         if parent_future is not None:
+            if parent_future.cancelled():
+                # The chain this event belongs to was cancelled; the event is
+                # stillborn and never enters the queue.
+                tracked.cancel()
+                return tracked
             parent_future.add_child(tracked)
+        self._supersede_previous(token=token, event=event, tracked=tracked)
         await queue.put(EventQueueEntry(event=event, ctx=ev_ctx))
         return tracked
 
@@ -491,14 +505,75 @@ def _try_clean_future(self, future: EventFuture) -> None:  # type: ignore[overri
         """
         if not future.done():
             return
+        if future.txid in self._tasks:
+            # The handler task is still running or unwinding; keep the future
+            # so late-chained events can find their (possibly cancelled) parent.
+            return
         # Not checking future.all_done() to avoid waiting for grandchildren here.
         if not all(c.done() for c in future.children):
             return
         parent = future.parent
         self._futures.pop(future.txid, None)
+        if (
+            (key := future.supersede_key) is not None
+            and self._superseded.get(key) is future
+            and future.all_done()
+        ):
+            del self._superseded[key]
         if parent is not None and parent.txid:
             self._try_clean_future(parent)
 
+    def _supersede_previous(
+        self, *, token: str, event: Event, tracked: EventFuture
+    ) -> None:
+        """Cancel the previous unfinished chain of a superseding event handler.
+
+        Handlers marked with ``supersedes`` (e.g. ``on_load_internal``) use
+        latest-wins semantics: enqueuing a new invocation cancels the previous
+        unfinished event chain for the same handler and client token.
+
+        Args:
+            token: The client token associated with the event.
+            event: The event being enqueued.
+            tracked: The future of the event being enqueued.
+        """
+        try:
+            registered = RegistrationContext.get().event_handlers.get(event.name)
+        except LookupError:
+            return
+        if registered is None or not registered.handler.supersedes:
+            return
+        key = (event.name, token)
+        previous = self._superseded.get(key)
+        if (
+            previous is not None
+            and not previous.all_done()
+            # A chain must not supersede itself (e.g. a handler re-enqueuing
+            # itself); cancelling an ancestor would cancel ``tracked`` too.
+            and not self._is_ancestor(previous, tracked)
+        ):
+            previous.cancel()
+        self._superseded[key] = tracked
+        tracked.supersede_key = key
+
+    @staticmethod
+    def _is_ancestor(candidate: EventFuture, future: EventFuture) -> bool:
+        """Check whether candidate is an ancestor of future.
+
+        Args:
+            candidate: The potential ancestor future.
+            future: The future whose parent chain to walk.
+
+        Returns:
+            True if candidate appears in future's parent chain.
+        """
+        parent = future.parent
+        while parent is not None:
+            if parent is candidate:
+                return True
+            parent = parent.parent
+        return False
+
     def _on_future_done(self, future: EventFuture) -> None:  # type: ignore[override]
         """Callback invoked when an enqueued future completes.
 
@@ -625,10 +700,13 @@ def _dispatch_next_for_token(self, token: str) -> None:
         if not token_queue:
             return
         entry, registered_handler = token_queue[0]
-        # Skip cancelled futures.
+        # Skip cancelled futures. Before a task exists, the only way a future
+        # can be done (and thus already cleaned up) is cancellation, so a
+        # missing future also means the entry was cancelled.
         future = self._futures.get(entry.ctx.txid)
-        if future is not None and future.cancelled():
-            self._try_clean_future(future)
+        if future is None or future.cancelled():
+            if future is not None:
+                self._try_clean_future(future)
             token_queue.popleft()
             if token_queue:
                 self._dispatch_next_for_token(token)
@@ -645,10 +723,12 @@ async def _process_queue(self):
         with contextlib.suppress(QueueShutDown):
             while True:
                 entry = await queue.get()
-                if (
-                    future := self._futures.get(entry.ctx.txid)
-                ) is not None and future.cancelled():
-                    self._try_clean_future(future)
+                # A missing future means the entry was cancelled and already
+                # cleaned up (see _dispatch_next_for_token).
+                future = self._futures.get(entry.ctx.txid)
+                if future is None or future.cancelled():
+                    if future is not None:
+                        self._try_clean_future(future)
                     queue.task_done()
                     continue
                 try:
@@ -757,6 +837,10 @@ def _finish_task(self, task: asyncio.Task):
             else:
                 if future is not None and not future.done():
                     future.set_result(result)
+        if future is not None:
+            # The task is gone; clean up now in case the future resolved
+            # earlier (e.g. external cancellation) and cleanup was deferred.
+            self._try_clean_future(future)
 
 
 __all__ = [

packages/reflex-base/src/reflex_base/event/processor/future.py

@@ -31,6 +31,10 @@ class EventFuture(asyncio.Future):
         default_factory=asyncio.get_running_loop, repr=False
     )
 
+    # Key under which this future is registered for latest-wins supersession
+    # in the EventProcessor, if any.
+    supersede_key: tuple[str, str] | None = dataclasses.field(default=None, repr=False)
+
     def __post_init__(self) -> None:
         """Call Future.__init__ for the EventFuture."""
         super(EventFuture, self).__init__(loop=self.loop)

reflex/state.py

@@ -32,6 +32,7 @@
 from reflex_base.event import (
     BACKGROUND_TASK_MARKER,
     EVENT_ACTIONS_MARKER,
+    SUPERSEDES_MARKER,
     Event,
     EventHandler,
     EventSpec,
@@ -739,6 +740,8 @@ def _copy_fn(fn: Callable) -> Callable:
         newfn.__annotations__ = fn.__annotations__
         if mark := getattr(fn, BACKGROUND_TASK_MARKER, None):
             setattr(newfn, BACKGROUND_TASK_MARKER, mark)
+        if mark := getattr(fn, SUPERSEDES_MARKER, None):
+            setattr(newfn, SUPERSEDES_MARKER, mark)
         # Preserve event_actions from @rx.event decorator
         if event_actions := getattr(fn, EVENT_ACTIONS_MARKER, None):
             object.__setattr__(newfn, EVENT_ACTIONS_MARKER, event_actions)
@@ -2464,6 +2467,15 @@ def on_load_internal(self) -> list[Event | EventSpec | event.EventCallback] | No
         ]
 
 
+# A newer navigation supersedes the previous unfinished on_load chain for the
+# same client token, cancelling its stale work (#6593).
+setattr(
+    OnLoadInternalState.event_handlers["on_load_internal"].fn,
+    SUPERSEDES_MARKER,
+    True,
+)
+
+
 class ComponentState(State, mixin=True):
     """Base class to allow for the creation of a state instance per component.
 

tests/integration/tests_playwright/test_router_query.py

@@ -13,6 +13,9 @@
   ``on_load``, and updates the router reactively with no further interaction.
 * ``rx.redirect(target, replace=True)`` behaves the same but replaces the
   current history entry instead of pushing a new one.
+* Navigating away while a page's ``on_load`` chain is still running cancels
+  the stale chain instead of letting it keep running and block the new
+  page's events (https://github.com/reflex-dev/reflex/issues/6593).
 
 Covers dev and prod modes via ``app_harness_env`` parametrisation.
 """
@@ -30,8 +33,27 @@
 
 def RouterQueryApp():
     """App exercising replaceState, redirect, and redirect(replace=True)."""
+    import asyncio
+
     import reflex as rx
 
+    class SlowLoadState(rx.State):
+        # Records the lifecycle of the slow on_load handler; "finished;" must
+        # never be appended when the user navigates away mid-load.
+        slow_log: str = ""
+
+        @rx.event
+        async def slow_on_load(self):
+            """Record start, simulate a slow data load, then record completion.
+
+            Yields:
+                None, to emit the "started;" delta before sleeping.
+            """
+            self.slow_log += "started;"
+            yield
+            await asyncio.sleep(2)
+            self.slow_log += "finished;"
+
     class RouterQueryState(rx.State):
         # Incremented by the page on_load handler; proves whether a navigation
         # (and thus on_load) actually fired.
@@ -146,10 +168,18 @@ def index():
                 read_only=True,
                 id="ping-count",
             ),
+            rx.input(value=SlowLoadState.slow_log, read_only=True, id="slow-log"),
+        )
+
+    def slow():
+        return rx.box(
+            rx.link("go home", href="/", id="to-index"),
+            rx.input(value=SlowLoadState.slow_log, read_only=True, id="slow-log"),
         )
 
     app = rx.App()
     app.add_page(index, route="/", on_load=RouterQueryState.on_load)
+    app.add_page(slow, route="/slow", on_load=SlowLoadState.slow_on_load)
 
 
 @pytest.fixture(scope="module")
@@ -275,3 +305,35 @@ def test_redirect_replace_replaces_history_entry(
     expect(page).to_have_url(f"{base}/")
     expect(page.locator("#name-param")).to_have_value("")
     expect(page.locator("#query-str")).to_have_value("")
+
+
+def test_slow_on_load_completes_without_navigation(
+    router_query_app: AppHarness, page: Page
+):
+    """Positive control: staying on the page lets the slow on_load finish."""
+    base = router_query_app.frontend_url
+    assert base is not None
+    page.goto(f"{base.rstrip('/')}/slow")
+    expect(page.locator("#slow-log")).to_have_value("started;")
+    expect(page.locator("#slow-log")).to_have_value("started;finished;", timeout=5000)
+
+
+def test_navigation_cancels_stale_on_load(router_query_app: AppHarness, page: Page):
+    """Navigating away mid-load cancels the stale on_load chain (#6593)."""
+    base = router_query_app.frontend_url
+    assert base is not None
+    page.goto(f"{base.rstrip('/')}/slow")
+    # The slow on_load has started and is now sleeping.
+    expect(page.locator("#slow-log")).to_have_value("started;")
+
+    # Navigate away while the slow handler is still running. The new page's
+    # on_load runs promptly instead of waiting ~2s behind the stale handler.
+    page.click("#to-index")
+    expect(page.locator("#load-count")).to_have_value("1", timeout=1000)
+
+    # Wait past the stale handler's sleep, then flush a round-trip; the
+    # cancelled handler must never have appended "finished;".
+    page.wait_for_timeout(3000)
+    page.click("#ping")
+    expect(page.locator("#ping-count")).to_have_value("1")
+    expect(page.locator("#slow-log")).to_have_value("started;")