ENG-9661 fix: cancel stale on_load chains via handler-declared supersession

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/news/6593.bugfix.md

@@ -0,0 +1 @@
+Event handlers marked with `@rx.event(supersedes=True)` now use latest-wins semantics: enqueuing a new invocation cancels the previous unfinished event chain for the same client token. `on_load_internal` uses this to cancel stale `on_load` chains on navigation.

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

@@ -291,6 +291,7 @@ def _scan_detach(value: Any, memo: dict[int, Any], active: set[int]) -> Any:
 
 
 BACKGROUND_TASK_MARKER = "_reflex_background_task"
+SUPERSEDES_MARKER = "_reflex_supersedes"
 EVENT_ACTIONS_MARKER = "_rx_event_actions"
 UPLOAD_FILES_CLIENT_HANDLER = "uploadFiles"
 
@@ -548,6 +549,19 @@ def is_background(self) -> bool:
         """
         return getattr(self.fn, BACKGROUND_TASK_MARKER, False)
 
+    @property
+    def supersedes(self) -> bool:
+        """Whether a newer chain-root invocation supersedes an older one.
+
+        When True, enqueuing this handler as a chain root 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.
 
@@ -2884,6 +2898,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
@@ -2909,6 +2924,7 @@ def __new__(
         func: None = None,
         *,
         background: bool | None = None,
+        supersedes: bool | None = None,
         stop_propagation: bool | None = None,
         prevent_default: bool | None = None,
         throttle: int | None = None,
@@ -2924,6 +2940,7 @@ def __new__(
         func: "Callable[[BASE_STATE, Unpack[P]], Any]",
         *,
         background: bool | None = None,
+        supersedes: bool | None = None,
         stop_propagation: bool | None = None,
         prevent_default: bool | None = None,
         throttle: int | None = None,
@@ -2936,6 +2953,7 @@ def __new__(
         func: "Callable[[BASE_STATE, Unpack[P]], Any] | None" = None,
         *,
         background: bool | None = None,
+        supersedes: bool | None = None,
         stop_propagation: bool | None = None,
         prevent_default: bool | None = None,
         throttle: int | None = None,
@@ -2947,6 +2965,9 @@ def __new__(
         Args:
             func: The function to wrap.
             background: Whether the event should be run in the background. Defaults to False.
+            supersedes: Whether enqueuing the event cancels the previous unfinished
+                chain of the same event for the same client token (latest-wins).
+                Defaults to False.
             stop_propagation: Whether to stop the event from bubbling up the DOM tree.
             prevent_default: Whether to prevent the default behavior of the event.
             throttle: Throttle the event handler to limit calls (in milliseconds).
@@ -2998,6 +3019,8 @@ def wrapper(
                     msg = "Background task must be async function or generator."
                     raise TypeError(msg)
                 setattr(func, BACKGROUND_TASK_MARKER, True)
+            if supersedes is True:
+                setattr(func, SUPERSEDES_MARKER, True)
             if getattr(func, "__name__", "").startswith("_"):
                 msg = "Event handlers cannot be private."
                 raise ValueError(msg)

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,14 @@ 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, so cancel the
+                # tracker since this event will never enter the queue.
+                tracked.cancel()
+                return tracked
             parent_future.add_child(tracked)
+        if parent_future is None:
+            self._supersede_previous(token=token, event=event, tracked=tracked)
         await queue.put(EventQueueEntry(event=event, ctx=ev_ctx))
         return tracked
 
@@ -491,14 +506,53 @@ def _try_clean_future(self, future: EventFuture) -> None:  # type: ignore[overri
         """
         if not future.done():
             return
+        if future.cancelled() and future.txid in self._tasks:
+            # The cancelled handler task is still unwinding; keep the future so
+            # late-chained events can find their cancelled parent. Failed
+            # futures are not retained, so a backend exception handler task
+            # reusing the txid can chain recovery events normally.
+            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.
+
+        Root 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():
+            previous.cancel()
+        self._superseded[key] = tracked
+        tracked.supersede_key = key
+
     def _on_future_done(self, future: EventFuture) -> None:  # type: ignore[override]
         """Callback invoked when an enqueued future completes.
 
@@ -625,10 +679,12 @@ 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 is cancellation, and its _try_clean_future done callback
+        # removes it from _futures, 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():
             token_queue.popleft()
             if token_queue:
                 self._dispatch_next_for_token(token)
@@ -645,10 +701,10 @@ 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():
                     queue.task_done()
                     continue
                 try:
@@ -757,6 +813,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

@@ -2444,6 +2444,9 @@ class OnLoadInternalState(State):
     # Cannot properly annotate this as `App` due to circular import issues.
     _app_ref: ClassVar[Any] = None
 
+    # A newer navigation supersedes the previous unfinished on_load chain for
+    # the same client token, cancelling its stale work (#6593).
+    @event(supersedes=True)
     def on_load_internal(self) -> list[Event | EventSpec | event.EventCallback] | None:
         """Queue on_load handlers for the current page.