fix(events): stop background tasks from cleaning the root state unlocked

news/6920.bugfix.md

@@ -0,0 +1 @@
+Fixed a race where a finishing background task could silently discard state updates made by a concurrently running event handler before they reached the frontend, leaving the UI stale until the next write. Background handlers that never enter `async with self` still emit their delta, now computed under the state lock.

packages/reflex-base/news/6920.bugfix.md

@@ -0,0 +1 @@
+Stop background event handlers from computing a delta and cleaning the root state after the state lock is dropped. On a shared state tree (opportunistic locking, in-memory state manager) that unlocked snapshot-then-clean raced concurrent foreground handlers: a write landing between the background task's dirty-var snapshot and its `_clean()` was silently discarded and never reached any delta. Background state changes are emitted by their `async with self` context exits, which hold the lock.

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

@@ -213,7 +213,12 @@ async def chain_updates(
     ctx = EventContext.get()
 
     if root_state is not None:
-        # Emit deltas first, so any frontend events are processed with the latest state.
+        # Emit deltas first, so any frontend events are processed with the
+        # latest state. The clean deliberately runs after resolution: the
+        # SharedState fan-out captures its dirty vars at clean time, and
+        # resolving the delta is what re-marks linked vars through the patch
+        # machinery, so cleaning earlier would fan out a stale set (see
+        # tests/integration/test_linked_state.py).
         try:
             delta = await root_state._get_resolved_delta()
             if delta:
@@ -228,19 +233,51 @@ async def chain_updates(
         await _route_events(ctx, fixed_events)
 
 
+def ensure_locked(
+    state: BaseState | StateProxy, root_state: BaseState | None
+) -> BaseState | None:
+    """The root to flush deltas from, only while the state lock is held.
+
+    Foreground handlers pass the locked root in. A background handler
+    yielding from inside ``async with self`` is suspended while its proxy
+    still holds the lock, so flushing through the proxy's root keeps the
+    documented ordering: deltas reach the frontend before the yielded event.
+    Outside the proxy context there is no lock, and flushing there is the
+    unlocked snapshot/clean that discards concurrent writes. Evaluated per
+    yield: a generator can move between inside and outside the context.
+
+    Args:
+        state: The state the handler runs against, possibly a StateProxy.
+        root_state: The locked root passed by foreground callers, if any.
+
+    Returns:
+        The root state to flush, or None when the lock is not held.
+    """
+    if root_state is not None:
+        return root_state
+    if isinstance(state, StateProxy) and state._is_mutable():
+        return state.__wrapped__._get_root_state()
+    return None
+
+
 async def process_event(
     handler: EventHandler,
     payload: dict,
     state: BaseState | StateProxy,
-    root_state: BaseState,
+    root_state: BaseState | None,
 ):
     """Process event.
 
     Args:
         handler: EventHandler to process.
         payload: The event payload.
         state: State to process the handler.
-        root_state: The root state of the app, used for emitting deltas.
+        root_state: The root state of the app, used for emitting deltas. Pass
+            None when the caller does not hold the state lock (background
+            tasks): computing and cleaning a delta on an unlocked root races
+            concurrent events on a shared state tree, and background state
+            changes are emitted by the ``async with self`` context exits
+            instead.
 
     Raises:
         ValueError: If a string value is received for an int or float type and cannot be converted.
@@ -269,28 +306,44 @@ async def process_event(
     # Handle async generators.
     if inspect.isasyncgen(events):
         async for event in events:
-            await chain_updates(event, root_state=root_state, handler_name=handler_name)
-        await chain_updates(None, root_state=root_state, handler_name=handler_name)
+            await chain_updates(
+                event,
+                root_state=ensure_locked(state, root_state),
+                handler_name=handler_name,
+            )
+        await chain_updates(
+            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
+        )
 
     # Handle regular generators.
     elif inspect.isgenerator(events):
         try:
             while True:
                 await chain_updates(
-                    next(events), root_state=root_state, handler_name=handler_name
+                    next(events),
+                    root_state=ensure_locked(state, root_state),
+                    handler_name=handler_name,
                 )
         except StopIteration as si:
             # the "return" value of the generator is not available
             # in the loop, we must catch StopIteration to access it
             if si.value is not None:
                 await chain_updates(
-                    si.value, root_state=root_state, handler_name=handler_name
+                    si.value,
+                    root_state=ensure_locked(state, root_state),
+                    handler_name=handler_name,
                 )
-        await chain_updates(None, root_state=root_state, handler_name=handler_name)
+        await chain_updates(
+            None, root_state=ensure_locked(state, root_state), handler_name=handler_name
+        )
 
     # Handle regular event chains.
     else:
-        await chain_updates(events, root_state=root_state, handler_name=handler_name)
+        await chain_updates(
+            events,
+            root_state=ensure_locked(state, root_state),
+            handler_name=handler_name,
+        )
 
 
 class BaseStateEventProcessor(EventProcessor):
@@ -395,13 +448,38 @@ async def _execute_event(
                     root_state=root_state,
                 )
                 return
-        # Otherwise drop the state lock and start processing the background task with a proxy state.
+        # Otherwise drop the state lock and start processing the background task
+        # with a proxy state. No root_state: the lock is no longer held, and
+        # under a shared state tree (opportunistic locking, in-memory manager)
+        # computing a delta here races whatever event holds the lock now -- a
+        # foreground write landing between this task's dirty-var snapshot and
+        # its _clean() would be discarded before any delta carries it. A
+        # background task's own state changes are emitted (and cleaned) by its
+        # `async with self` context exits, which re-acquire the lock.
+        proxy = StateProxy(substate)
         await process_event(
             handler=registered_handler.handler,
-            state=StateProxy(substate),
+            state=proxy,
             payload=event.payload,
-            root_state=root_state,
+            root_state=None,
         )
+        if not proxy._self_entered_context:
+            # A handler that never entered `async with self` emitted nothing,
+            # but every background event used to flush a delta (refreshing
+            # uncached computed vars, and any dirty vars the preamble left,
+            # like router_data). Preserve that, under the lock this time.
+            async with ctx.state_manager.modify_state_with_links(
+                BaseStateToken(
+                    ident=ctx.token,
+                    cls=registered_handler.states[0],
+                ),
+                event=event,
+            ) as flush_state:
+                await chain_updates(
+                    None,
+                    root_state=flush_state._get_root_state(),
+                    handler_name=registered_handler.handler.fn.__qualname__,
+                )
 
     async def _handle_backend_exception(
         self, ex: Exception, ev_ctx: EventContext | None = None

reflex/istate/proxy.py

@@ -96,6 +96,9 @@ def __init__(
         self._self_actx_lock = asyncio.Lock()
         self._self_actx_lock_holder = None
         self._self_parent_state_proxy = parent_state_proxy
+        # Whether `async with self` was ever entered; a background handler that
+        # never did emitted no delta, so the processor flushes once for it.
+        self._self_entered_context = False
 
     def _is_mutable(self) -> bool:
         """Check if the state is mutable.
@@ -134,6 +137,7 @@ async def __aenter__(self) -> Self:
                     State.get_class_substate(self._self_substate_path)
                 ),
             )
+            self._self_entered_context = True
             return self
         current_task = asyncio.current_task()
         if (
@@ -153,6 +157,7 @@ async def __aenter__(self) -> Self:
             )
             mutable_state = await self._self_actx.__aenter__()
             self._self_mutable = True
+            self._self_entered_context = True
             super().__setattr__(
                 "__wrapped__", mutable_state.get_substate(self._self_substate_path)
             )