Only propagate newly-dirty vars in _mark_dirty_computed_vars

news/6743.performance.md

@@ -0,0 +1 @@
+Dirty propagation now only walks newly-dirty vars per mutation and skips the computed var expiry scan for classes with no interval vars, roughly halving per-mutation overhead in states with computed vars.

packages/reflex-base/news/6743.performance.md

@@ -0,0 +1 @@
+Dirty propagation now only walks newly-dirty vars per mutation and skips the computed var expiry scan for classes with no interval vars, roughly halving per-mutation overhead in states with computed vars.

packages/reflex-base/src/reflex_base/vars/base.py

@@ -2170,6 +2170,19 @@ def is_computed_var(obj: Any) -> TypeGuard[ComputedVar]:
     return isinstance(obj, FakeComputedVarBaseClass)
 
 
+# Incremented whenever a cached computed var is recomputed. State dirty
+# propagation uses this to know when an already-propagated dependency needs to
+# be re-propagated (a recompute re-materializes a cache that a later mutation
+# of its dependencies must invalidate again).
+_computed_var_recompute_generation: int = 0
+
+
+def _bump_computed_var_recompute_generation() -> None:
+    """Record that a cached computed var was recomputed."""
+    global _computed_var_recompute_generation
+    _computed_var_recompute_generation += 1
+
+
 @dataclasses.dataclass(
     eq=False,
     frozen=True,
@@ -2511,6 +2524,7 @@ def __get__(self, instance: BaseState | None, owner: type):
                 instance._was_touched = True
                 # Set the last updated timestamp on the state instance.
                 setattr(instance, self._last_updated_attr, datetime.datetime.now())
+                _bump_computed_var_recompute_generation()
             value = getattr(instance, self._cache_attr)
 
         self._check_deprecated_return_type(instance, value)
@@ -2773,6 +2787,7 @@ async def _awaitable_result(instance: BaseState = instance) -> RETURN_TYPE:
                 instance._was_touched = True
                 # Set the last updated timestamp on the state instance.
                 setattr(instance, self._last_updated_attr, datetime.datetime.now())
+                _bump_computed_var_recompute_generation()
             value = getattr(instance, self._cache_attr)
             self._check_deprecated_return_type(instance, value)
             return value

reflex/state.py

@@ -56,6 +56,7 @@
 from reflex_base.utils.serializers import serializer
 from reflex_base.utils.types import _isinstance
 from reflex_base.vars import Field, VarData, field
+from reflex_base.vars import base as reflex_base_vars_base
 from reflex_base.vars.base import (
     ComputedVar,
     DynamicRouteVar,
@@ -369,6 +370,7 @@ def _is_user_descriptor(value: Any) -> bool:
     "_always_dirty_computed_vars",
     "_always_dirty_substates",
     "_potentially_dirty_states",
+    "_interval_computed_vars",
 })
 
 
@@ -408,6 +410,11 @@ class BaseState(EvenMoreBasicBaseState):
     # Set of states which might need to be recomputed if vars in this state change.
     _potentially_dirty_states: ClassVar[set[str]] = set()
 
+    # Names of computed vars with an update interval, refreshed whenever
+    # computed_vars changes. Empty for most classes, which lets dirty
+    # propagation skip the per-mutation expiry scan.
+    _interval_computed_vars: ClassVar[tuple[str, ...]] = ()
+
     # The parent state.
     parent_state: BaseState | None = field(default=None, is_var=False)
 
@@ -718,9 +725,22 @@ def __init_subclass__(cls, mixin: bool = False, **kwargs):
         # Initialize per-class var dependency tracking.
         cls._var_dependencies = {}
         cls._init_var_dependency_dicts()
+        cls._refresh_interval_computed_vars()
 
         all_base_state_classes[cls.get_full_name()] = None
 
+    @classmethod
+    def _refresh_interval_computed_vars(cls) -> None:
+        """Recompute the names of computed vars that have an update interval.
+
+        Must be called whenever cls.computed_vars is modified.
+        """
+        cls._interval_computed_vars = tuple(
+            name
+            for name, cvar in cls.computed_vars.items()
+            if cvar._update_interval is not None
+        )
+
     @classmethod
     def _add_event_handler(
         cls,
@@ -828,6 +848,7 @@ def computed_var_func(state: Self):
 
         setattr(cls, unique_var_name, computed_var_func_arg)
         cls.computed_vars[unique_var_name] = computed_var_func_arg
+        cls._refresh_interval_computed_vars()
         cls.vars[unique_var_name] = computed_var_func_arg
         cls._update_substate_inherited_vars({unique_var_name: computed_var_func_arg})
         cls._always_dirty_computed_vars.add(unique_var_name)
@@ -1401,6 +1422,7 @@ def inner_func(self: BaseState) -> list[str]:
 
         # Update tracking dicts.
         cls.computed_vars.update(dynamic_vars)
+        cls._refresh_interval_computed_vars()
         cls.vars.update(dynamic_vars)
         cls._update_substate_inherited_vars(dynamic_vars)
 
@@ -1798,14 +1820,31 @@ async def get_var_value(self, var: Var[VAR_TYPE]) -> VAR_TYPE:
 
     def _mark_dirty_computed_vars(self) -> None:
         """Mark ComputedVars that need to be recalculated based on dirty_vars."""
-        # Append expired computed vars to dirty_vars to trigger recalculation
-        self.dirty_vars.update(self._expired_computed_vars())
+        if self._interval_computed_vars:
+            # Append expired computed vars to dirty_vars to trigger recalculation
+            self.dirty_vars.update(self._expired_computed_vars())
         # Append always dirty computed vars to dirty_vars to trigger recalculation
         self.dirty_vars.update(self._always_dirty_computed_vars)
 
-        dirty_vars = self.dirty_vars
-        while dirty_vars:
-            calc_vars, dirty_vars = dirty_vars, set()
+        # Track which dirty vars already had their dependency closure
+        # propagated, so repeated mutations only process newly-dirty vars.
+        # Recomputing any cached var re-materializes a cache that a later
+        # mutation must invalidate again, so the frontier is only valid for
+        # the recompute generation it was built in.
+        instance_dict = self.__dict__
+        propagated = instance_dict.get("_propagated_dirty_vars")
+        current_gen = reflex_base_vars_base._computed_var_recompute_generation
+        if propagated is None:
+            propagated = set()
+            object.__setattr__(self, "_propagated_dirty_vars", propagated)
+        elif instance_dict.get("_propagated_generation") != current_gen:
+            propagated.clear()
+        object.__setattr__(self, "_propagated_generation", current_gen)
+
+        new_dirty = self.dirty_vars - propagated
+        while new_dirty:
+            propagated |= new_dirty
+            calc_vars, new_dirty = new_dirty, set()
             for state_name, cvar in self._dirty_computed_vars(from_vars=calc_vars):
                 if state_name == self.get_full_name():
                     defining_state = self
@@ -1818,7 +1857,8 @@ def _mark_dirty_computed_vars(self) -> None:
                 if actual_var is not None:
                     actual_var.mark_dirty(instance=defining_state)
                 if defining_state is self:
-                    dirty_vars.add(cvar)
+                    if cvar not in propagated:
+                        new_dirty.add(cvar)
                 else:
                     # mark dirty where this var is defined
                     defining_state._mark_dirty()
@@ -1829,10 +1869,11 @@ def _expired_computed_vars(self) -> set[str]:
         Returns:
             Set of computed vars to include in the delta.
         """
+        computed_vars = self.computed_vars
         return {
             cvar
-            for cvar, cvar_obj in self.computed_vars.items()
-            if cvar_obj.needs_update(instance=self)
+            for cvar in self._interval_computed_vars
+            if computed_vars[cvar].needs_update(instance=self)
         }
 
     def _dirty_computed_vars(
@@ -1952,6 +1993,8 @@ def _clean(self):
         # Clean this state.
         self.dirty_vars = set()
         self.dirty_substates = set()
+        # Discard the propagation frontier along with the dirty vars it tracked.
+        self.__dict__.pop("_propagated_dirty_vars", None)
 
     def get_value(self, key: str) -> Any:
         """Get the value of a field (without proxying).
@@ -2069,6 +2112,9 @@ def __getstate__(self):
         state.pop("parent_state", None)
         state.pop("substates", None)
         state.pop("_was_touched", None)
+        # The propagation frontier is transient and rebuilt on demand.
+        state.pop("_propagated_dirty_vars", None)
+        state.pop("_propagated_generation", None)
         # Remove all inherited vars.
         for inherited_var_name in self.inherited_vars:
             state.pop(inherited_var_name, None)

tests/units/test_state.py

@@ -1437,6 +1437,63 @@ def comp_v(self) -> int:
     assert comp_v_calls == 2
 
 
+def test_computed_var_recompute_after_mid_cycle_read():
+    """A dependency mutated again after a mid-cycle read still invalidates the cache."""
+
+    class FrontierState(BaseState):
+        v: int = 0
+
+        @rx.var
+        def doubled(self) -> int:
+            return self.v * 2
+
+    s = FrontierState()
+    s.v = 1
+    # Reading mid-cycle recomputes and re-caches the value.
+    assert s.doubled == 2
+    # Mutating the same dependency again must invalidate the fresh cache,
+    # even though dirty_vars already contained it.
+    s.v = 2
+    assert s.doubled == 4
+    assert s.get_delta()[s.get_full_name()]["doubled" + FIELD_MARKER] == 4
+
+
+def test_computed_var_recompute_after_mid_cycle_read_across_states():
+    """Cross-state dependency invalidation survives a mid-cycle recompute."""
+
+    class FrontierParentState(BaseState):
+        v: int = 0
+
+    class FrontierChildState(FrontierParentState):
+        @rx.var
+        def doubled(self) -> int:
+            return self.v * 2
+
+    parent = FrontierParentState()
+    child = parent.substates[FrontierChildState.get_name()]
+    parent.v = 1
+    assert child.doubled == 2
+    parent.v = 2
+    assert child.doubled == 4
+
+
+def test_interval_computed_vars_precomputed():
+    """Classes precompute which computed vars carry an update interval."""
+
+    class IntervalFreeState(BaseState):
+        @rx.var
+        def untimed(self) -> int:
+            return 2
+
+    class IntervalState(BaseState):
+        @rx.var(interval=15)
+        def timed(self) -> int:
+            return 1
+
+    assert IntervalFreeState._interval_computed_vars == ()
+    assert IntervalState._interval_computed_vars == ("timed",)
+
+
 def test_computed_var_cached_depends_on_non_cached():
     """Test that a cached var is recalculated if it depends on non-cached ComputedVar."""