From 148da1efe2bb7db4987385f745777596df4f4fcd Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 4 Aug 2026 16:28:48 +0530 Subject: [PATCH 01/19] Avoid root dirty delta for shared state events --- news/6392.bugfix.md | 1 + reflex/istate/shared.py | 3 +++ tests/units/test_state.py | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 39 insertions(+) create mode 100644 news/6392.bugfix.md diff --git a/news/6392.bugfix.md b/news/6392.bugfix.md new file mode 100644 index 00000000000..755e5824602 --- /dev/null +++ b/news/6392.bugfix.md @@ -0,0 +1 @@ +Avoid leaking temporary router dirtiness into linked/shared state event deltas. diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 9e928caea3f..7a3b2a665e8 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -115,6 +115,9 @@ async def _patch_state( linked_state.dirty_vars.update(linked_state.computed_vars) linked_state._mark_dirty() # Apply the updates into the existing state tree for rehydrate. + # For regular linked-state events this router dirtiness is temporary: + # it forces router-dependent computed vars to resolve for the patched + # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) diff --git a/tests/units/test_state.py b/tests/units/test_state.py index d0c1192779e..ea9e212843b 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -919,6 +919,41 @@ def test_setting_inherited_backend_var_does_not_mark_child_touched( assert not child_touched +class _LinkedStatePatchRoot(BaseState): + """Root state for testing linked-state dirty propagation.""" + + +class _LinkedStatePatchShared(_LinkedStatePatchRoot): + """Substate used to exercise _patch_state without full SharedState setup.""" + + counter: int = 0 + + +@pytest.mark.asyncio +async def test_linked_state_event_does_not_dirty_root_state(): + """Linked-state events should not leak temporary router dirtiness.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + linked_state.counter = 1 + + assert "router" not in private_tree.dirty_vars + assert constants.ROUTER_DATA not in private_tree.dirty_vars + assert private_tree.get_full_name() not in private_tree.get_delta() + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From eac81b38cfac3e107366eaba9515711f18c6f219 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 4 Aug 2026 16:55:16 +0530 Subject: [PATCH 02/19] Handle linked state patch resolve failures --- news/{6392.bugfix.md => 6841.bugfix.md} | 0 reflex/istate/shared.py | 15 ++++++--- tests/units/test_state.py | 41 +++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 4 deletions(-) rename news/{6392.bugfix.md => 6841.bugfix.md} (100%) diff --git a/news/6392.bugfix.md b/news/6841.bugfix.md similarity index 100% rename from news/6392.bugfix.md rename to news/6841.bugfix.md diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 7a3b2a665e8..eebb3182f92 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -119,13 +119,20 @@ async def _patch_state( # it forces router-dependent computed vars to resolve for the patched # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() + root_dirty_vars = set(root_state.dirty_vars) + root_dirty_substates = set(root_state.dirty_substates) root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() - # The delta is discarded: it is only resolved to refresh computed vars, - # so its values must not count as sent to the client. - with _suppress_delta_recording(): - await root_state._get_resolved_delta() + try: + # The delta is discarded: it is only resolved to refresh computed vars, + # so its values must not count as sent to the client. + with _suppress_delta_recording(): + await root_state._get_resolved_delta() + finally: + if not full_delta: + root_state.dirty_vars = root_dirty_vars + root_state.dirty_substates = root_dirty_substates yield finally: original_parent_state.substates[state_name] = original_state diff --git a/tests/units/test_state.py b/tests/units/test_state.py index ea9e212843b..28e43088ad0 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -922,6 +922,8 @@ def test_setting_inherited_backend_var_does_not_mark_child_touched( class _LinkedStatePatchRoot(BaseState): """Root state for testing linked-state dirty propagation.""" + value: int = 0 + class _LinkedStatePatchShared(_LinkedStatePatchRoot): """Substate used to exercise _patch_state without full SharedState setup.""" @@ -954,6 +956,45 @@ async def test_linked_state_event_does_not_dirty_root_state(): assert private_tree.get_full_name() not in private_tree.get_delta() +@pytest.mark.asyncio +async def test_linked_state_patch_restores_root_dirty_state_on_resolve_error( + monkeypatch: pytest.MonkeyPatch, +): + """Temporary root dirtiness should be cleaned if delta resolution fails.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree.value = 1 + private_tree.dirty_substates.add("existing") + original_dirty_vars = set(private_tree.dirty_vars) + original_dirty_substates = set(private_tree.dirty_substates) + + async def raise_resolve_error(): + await asyncio.sleep(0) + msg = "delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert private_tree.dirty_vars == original_dirty_vars + assert private_tree.dirty_substates == original_dirty_substates + assert private_tree.substates[shared_state_name] is private_state + assert linked_state.parent_state is linked_tree + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From e90f8f47adcdb46fe53242ad13ded428bbf1b57d Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 8 Aug 2026 10:04:53 +0530 Subject: [PATCH 03/19] Restore descendant dirty state on linked patch errors --- reflex/istate/shared.py | 17 +++++++++++++++++ tests/units/test_state.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index eebb3182f92..4f04c842c71 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -121,6 +121,17 @@ async def _patch_state( root_state = original_state._get_root_state() root_dirty_vars = set(root_state.dirty_vars) root_dirty_substates = set(root_state.dirty_substates) + dirty_state_snapshots: list[tuple[BaseState, set[str], set[str]]] = [] + if not full_delta: + states_to_snapshot = [root_state] + while states_to_snapshot: + state = states_to_snapshot.pop() + dirty_state_snapshots.append(( + state, + set(state.dirty_vars), + set(state.dirty_substates), + )) + states_to_snapshot.extend(state.substates.values()) root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() @@ -129,6 +140,12 @@ async def _patch_state( # so its values must not count as sent to the client. with _suppress_delta_recording(): await root_state._get_resolved_delta() + except BaseException: + if not full_delta: + for state, dirty_vars, dirty_substates in dirty_state_snapshots: + state.dirty_vars = dirty_vars + state.dirty_substates = dirty_substates + raise finally: if not full_delta: root_state.dirty_vars = root_dirty_vars diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 28e43088ad0..ca335159278 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -995,6 +995,36 @@ async def raise_resolve_error(): assert linked_state.parent_state is linked_tree +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_on_resolve_error(): + """Temporary descendant dirtiness should be cleaned on resolution failure.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def raise_resolve_error(): + await asyncio.sleep(0) + linked_state.dirty_vars.add("temporary") + linked_state._mark_dirty() + msg = "descendant delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="descendant delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From 1ca1067a8d785bf148ccda93aad8c3e46c7a24d3 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 13 Aug 2026 09:38:36 +0530 Subject: [PATCH 04/19] Restore descendant dirty state after shared patch --- reflex/istate/shared.py | 9 +-------- tests/units/test_state.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 8 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 4f04c842c71..1e9d6b997a2 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -119,8 +119,6 @@ async def _patch_state( # it forces router-dependent computed vars to resolve for the patched # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() - root_dirty_vars = set(root_state.dirty_vars) - root_dirty_substates = set(root_state.dirty_substates) dirty_state_snapshots: list[tuple[BaseState, set[str], set[str]]] = [] if not full_delta: states_to_snapshot = [root_state] @@ -140,16 +138,11 @@ async def _patch_state( # so its values must not count as sent to the client. with _suppress_delta_recording(): await root_state._get_resolved_delta() - except BaseException: + finally: if not full_delta: for state, dirty_vars, dirty_substates in dirty_state_snapshots: state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates - raise - finally: - if not full_delta: - root_state.dirty_vars = root_dirty_vars - root_state.dirty_substates = root_dirty_substates yield finally: original_parent_state.substates[state_name] = original_state diff --git a/tests/units/test_state.py b/tests/units/test_state.py index ca335159278..fcfa4d1a8d9 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -1025,6 +1025,36 @@ async def raise_resolve_error(): assert private_tree.dirty_substates == set() +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_after_resolve(): + """Temporary descendant dirtiness should be cleaned after resolution.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def resolve_with_temporary_dirty_state(): + await asyncio.sleep(0) + linked_state.dirty_vars.add("temporary") + linked_state._mark_dirty() + return {} + + object.__setattr__( + private_tree, "_get_resolved_delta", resolve_with_temporary_dirty_state + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From 561b2c306d3a95c390e35c3e297bc273eaf46c7a Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 15 Aug 2026 08:14:38 +0530 Subject: [PATCH 05/19] Preserve computed refreshes during shared state patching --- reflex/istate/shared.py | 75 +++++++++++++++++++++++++++++++++++++-- tests/units/test_state.py | 40 +++++++++++++++++++++ 2 files changed, 112 insertions(+), 3 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 1e9d6b997a2..75f1520c217 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -119,30 +119,99 @@ async def _patch_state( # it forces router-dependent computed vars to resolve for the patched # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() - dirty_state_snapshots: list[tuple[BaseState, set[str], set[str]]] = [] + dirty_state_snapshots: list[tuple[BaseState, set[str], set[str], set[str]]] = [] if not full_delta: states_to_snapshot = [root_state] while states_to_snapshot: state = states_to_snapshot.pop() + # Interval and always-dirty computed vars are valid output from + # this refresh and must not be mistaken for temporary router dirtiness. + computed_vars_to_preserve = state._expired_computed_vars().union( + state._always_dirty_computed_vars + ) dirty_state_snapshots.append(( state, set(state.dirty_vars), set(state.dirty_substates), + computed_vars_to_preserve, )) states_to_snapshot.extend(state.substates.values()) root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() + router_dirty_snapshots: list[ + tuple[BaseState, set[str], set[str], set[str], set[str], set[str]] + ] = [] + if not full_delta: + for ( + state, + dirty_vars, + dirty_substates, + computed_vars_to_preserve, + ) in dirty_state_snapshots: + router_dirty_snapshots.append(( + state, + dirty_vars, + dirty_substates, + state.dirty_vars - dirty_vars - computed_vars_to_preserve, + state.dirty_substates - dirty_substates, + computed_vars_to_preserve, + )) try: # The delta is discarded: it is only resolved to refresh computed vars, # so its values must not count as sent to the client. with _suppress_delta_recording(): await root_state._get_resolved_delta() - finally: + except BaseException: if not full_delta: - for state, dirty_vars, dirty_substates in dirty_state_snapshots: + for state, dirty_vars, dirty_substates, _ in dirty_state_snapshots: state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates + raise + else: + if not full_delta: + computed_refresh_state_names = { + state.get_full_name() + for ( + state, + _, + _, + router_dirty_vars, + _, + computed_vars_to_preserve, + ) in router_dirty_snapshots + if computed_vars_to_preserve + or (state.dirty_vars - router_dirty_vars).intersection( + state.computed_vars + ) + } + for ( + state, + dirty_vars, + dirty_substates, + router_dirty_vars, + router_dirty_substates, + computed_vars_to_preserve, + ) in router_dirty_snapshots: + computed_vars_refreshed = ( + state.dirty_vars - router_dirty_vars + ).intersection(state.computed_vars) + substates_with_computed_refresh = { + substate_name + for substate_name in state.dirty_substates + - router_dirty_substates + if any( + refresh_name == substate_name + or refresh_name.startswith(substate_name + ".") + for refresh_name in computed_refresh_state_names + ) + } + state.dirty_vars = ( + dirty_vars | computed_vars_to_preserve | computed_vars_refreshed + ) + state.dirty_substates = dirty_substates | ( + substates_with_computed_refresh + ) yield finally: original_parent_state.substates[state_name] = original_state diff --git a/tests/units/test_state.py b/tests/units/test_state.py index fcfa4d1a8d9..34e46d4fe78 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -931,6 +931,22 @@ class _LinkedStatePatchShared(_LinkedStatePatchRoot): counter: int = 0 +class _LinkedStatePatchIntervalRoot(BaseState): + """Root state for testing interval computed refreshes during patching.""" + + value: int = 0 + + @rx.var(interval=60) + def interval_value(self) -> int: + return self.value + + +class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): + """Substate used to exercise interval computed refreshes.""" + + counter: int = 0 + + @pytest.mark.asyncio async def test_linked_state_event_does_not_dirty_root_state(): """Linked-state events should not leak temporary router dirtiness.""" @@ -1055,6 +1071,30 @@ async def resolve_with_temporary_dirty_state(): assert private_tree.dirty_substates == set() +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_interval_computed_refresh(): + """Interval computed vars refreshed during patching must be emitted.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert "interval_value" in private_tree.dirty_vars + assert ( + "interval_value" + FIELD_MARKER + in private_tree.get_delta()[private_tree.get_full_name()] + ) + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From 779d62e797ef9acb5d0daaec4da5d903ac749b8e Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 15 Aug 2026 09:22:25 +0530 Subject: [PATCH 06/19] Preserve linked computed refresh paths --- reflex/istate/shared.py | 78 +++++++++++++++++++++++---------------- tests/units/test_state.py | 65 +++++++++++++++++++++++++++----- 2 files changed, 102 insertions(+), 41 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 75f1520c217..00cb8d81d44 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -119,7 +119,11 @@ async def _patch_state( # it forces router-dependent computed vars to resolve for the patched # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() - dirty_state_snapshots: list[tuple[BaseState, set[str], set[str], set[str]]] = [] + dirty_state_snapshots: list[ + tuple[ + BaseState, set[str], set[str], set[str], dict[str, tuple[bool, object]] + ] + ] = [] if not full_delta: states_to_snapshot = [root_state] while states_to_snapshot: @@ -129,18 +133,34 @@ async def _patch_state( computed_vars_to_preserve = state._expired_computed_vars().union( state._always_dirty_computed_vars ) + computed_var_snapshots = { + name: ( + hasattr(state, computed_var._cache_attr), + getattr(state, computed_var._last_updated_attr, None), + ) + for name, computed_var in state.computed_vars.items() + } dirty_state_snapshots.append(( state, set(state.dirty_vars), set(state.dirty_substates), computed_vars_to_preserve, + computed_var_snapshots, )) states_to_snapshot.extend(state.substates.values()) root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() router_dirty_snapshots: list[ - tuple[BaseState, set[str], set[str], set[str], set[str], set[str]] + tuple[ + BaseState, + set[str], + set[str], + set[str], + set[str], + set[str], + dict[str, tuple[bool, object]], + ] ] = [] if not full_delta: for ( @@ -148,6 +168,7 @@ async def _patch_state( dirty_vars, dirty_substates, computed_vars_to_preserve, + computed_var_snapshots, ) in dirty_state_snapshots: router_dirty_snapshots.append(( state, @@ -156,6 +177,7 @@ async def _patch_state( state.dirty_vars - dirty_vars - computed_vars_to_preserve, state.dirty_substates - dirty_substates, computed_vars_to_preserve, + computed_var_snapshots, )) try: # The delta is discarded: it is only resolved to refresh computed vars, @@ -164,54 +186,46 @@ async def _patch_state( await root_state._get_resolved_delta() except BaseException: if not full_delta: - for state, dirty_vars, dirty_substates, _ in dirty_state_snapshots: + for state, dirty_vars, dirty_substates, _, _ in dirty_state_snapshots: state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates raise else: if not full_delta: - computed_refresh_state_names = { - state.get_full_name() - for ( - state, - _, - _, - router_dirty_vars, - _, - computed_vars_to_preserve, - ) in router_dirty_snapshots - if computed_vars_to_preserve - or (state.dirty_vars - router_dirty_vars).intersection( - state.computed_vars - ) - } + computed_refresh_states: list[BaseState] = [] for ( state, dirty_vars, dirty_substates, router_dirty_vars, - router_dirty_substates, + _, computed_vars_to_preserve, + computed_var_snapshots, ) in router_dirty_snapshots: - computed_vars_refreshed = ( + computed_vars_refreshed = set(computed_vars_to_preserve) | ( state.dirty_vars - router_dirty_vars ).intersection(state.computed_vars) - substates_with_computed_refresh = { - substate_name - for substate_name in state.dirty_substates - - router_dirty_substates - if any( - refresh_name == substate_name - or refresh_name.startswith(substate_name + ".") - for refresh_name in computed_refresh_state_names + computed_vars_refreshed.update( + name + for name, computed_var in state.computed_vars.items() + if computed_var_snapshots[name] + != ( + hasattr(state, computed_var._cache_attr), + getattr(state, computed_var._last_updated_attr, None), ) - } + and hasattr(state, computed_var._cache_attr) + ) + if computed_vars_refreshed: + computed_refresh_states.append(state) state.dirty_vars = ( dirty_vars | computed_vars_to_preserve | computed_vars_refreshed ) - state.dirty_substates = dirty_substates | ( - substates_with_computed_refresh - ) + state.dirty_substates = dirty_substates + for refreshed_state in computed_refresh_states: + state = refreshed_state + while state.parent_state is not None: + state.parent_state.dirty_substates.add(state.get_name()) + state = state.parent_state yield finally: original_parent_state.substates[state_name] = original_state diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 34e46d4fe78..59cd4896d96 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -930,22 +930,26 @@ class _LinkedStatePatchShared(_LinkedStatePatchRoot): counter: int = 0 + @rx.var + def root_value(self) -> int: + return self.value + class _LinkedStatePatchIntervalRoot(BaseState): """Root state for testing interval computed refreshes during patching.""" value: int = 0 - @rx.var(interval=60) - def interval_value(self) -> int: - return self.value - class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): """Substate used to exercise interval computed refreshes.""" counter: int = 0 + @rx.var(interval=60) + def interval_value(self) -> int: + return self.value + @pytest.mark.asyncio async def test_linked_state_event_does_not_dirty_root_state(): @@ -1064,7 +1068,7 @@ async def resolve_with_temporary_dirty_state(): ) async with _patch_state(private_state, linked_state, full_delta=False): - pass + assert private_tree.substates[shared_state_name] is linked_state assert linked_state.dirty_vars == set() assert linked_state.dirty_substates == set() @@ -1085,13 +1089,56 @@ async def test_linked_state_patch_preserves_interval_computed_refresh(): private_tree._clean() + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_dirty_child(): + linked_state._mark_dirty() + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_dirty_child) + async with _patch_state(private_state, linked_state, full_delta=False): - pass + assert private_tree.substates[shared_state_name] is linked_state + interval_delta = private_tree.get_delta() + + assert "interval_value" in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} + assert ( + "interval_value" + FIELD_MARKER in interval_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): + """Computed vars invalidated through the root must remain in the delta.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_root_change(): + private_tree.value = 1 + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_root_change) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + root_dependent_delta = private_tree.get_delta() - assert "interval_value" in private_tree.dirty_vars + assert "root_value" in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} assert ( - "interval_value" + FIELD_MARKER - in private_tree.get_delta()[private_tree.get_full_name()] + "root_value" + FIELD_MARKER + in root_dependent_delta[linked_state.get_full_name()] ) From 8f9548b2ebd6660a4a4d096bded2f30d84ffdbc8 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 15 Aug 2026 10:18:55 +0530 Subject: [PATCH 07/19] Restore computed caches after shared patch failure --- reflex/istate/shared.py | 41 ++++++++++++++++++++++++++++++++--- tests/units/test_state.py | 45 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 3 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 00cb8d81d44..2139a45d0a8 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -121,7 +121,11 @@ async def _patch_state( root_state = original_state._get_root_state() dirty_state_snapshots: list[ tuple[ - BaseState, set[str], set[str], set[str], dict[str, tuple[bool, object]] + BaseState, + set[str], + set[str], + set[str], + dict[str, tuple[bool, object, bool, object]], ] ] = [] if not full_delta: @@ -136,6 +140,8 @@ async def _patch_state( computed_var_snapshots = { name: ( hasattr(state, computed_var._cache_attr), + getattr(state, computed_var._cache_attr, None), + hasattr(state, computed_var._last_updated_attr), getattr(state, computed_var._last_updated_attr, None), ) for name, computed_var in state.computed_vars.items() @@ -159,7 +165,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, object]], + dict[str, tuple[bool, object, bool, object]], ] ] = [] if not full_delta: @@ -186,9 +192,36 @@ async def _patch_state( await root_state._get_resolved_delta() except BaseException: if not full_delta: - for state, dirty_vars, dirty_substates, _, _ in dirty_state_snapshots: + for ( + state, + dirty_vars, + dirty_substates, + _, + computed_var_snapshots, + ) in dirty_state_snapshots: state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates + for name, computed_var in state.computed_vars.items(): + ( + had_cache, + cached_value, + had_last_updated, + last_updated, + ) = computed_var_snapshots[name] + if had_cache: + setattr(state, computed_var._cache_attr, cached_value) + else: + with contextlib.suppress(AttributeError): + delattr(state, computed_var._cache_attr) + if had_last_updated: + setattr( + state, + computed_var._last_updated_attr, + last_updated, + ) + else: + with contextlib.suppress(AttributeError): + delattr(state, computed_var._last_updated_attr) raise else: if not full_delta: @@ -211,6 +244,8 @@ async def _patch_state( if computed_var_snapshots[name] != ( hasattr(state, computed_var._cache_attr), + getattr(state, computed_var._cache_attr, None), + hasattr(state, computed_var._last_updated_attr), getattr(state, computed_var._last_updated_attr, None), ) and hasattr(state, computed_var._cache_attr) diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 59cd4896d96..c2c77ad681d 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -1142,6 +1142,51 @@ async def resolve_with_root_change(): ) +@pytest.mark.asyncio +async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): + """Failed resolution should not hide a partially refreshed computed value.""" + from reflex.istate.shared import _patch_state + + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + computed_var = _LinkedStatePatchIntervalShared.computed_vars["interval_value"] + cache_attr = computed_var._cache_attr + last_updated_attr = computed_var._last_updated_attr + cache_before = ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) + + private_tree._clean() + resolve_delta = private_tree._get_resolved_delta + + async def resolve_then_fail(): + linked_state._mark_dirty() + await resolve_delta() + msg = "computed refresh failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_then_fail) + + with pytest.raises(RuntimeError, match="computed refresh failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + assert linked_state.dirty_vars == set() + assert ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) == cache_before + + @pytest.mark.asyncio async def test_process_event_simple( token: str, From 1ada24df7858118ac1d012b6a42150c9e8153906 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 15 Aug 2026 12:48:56 +0530 Subject: [PATCH 08/19] Restore touched state after shared patch failure --- reflex/istate/shared.py | 5 +++++ tests/units/test_state.py | 2 ++ 2 files changed, 7 insertions(+) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 2139a45d0a8..16160daae1e 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -126,6 +126,7 @@ async def _patch_state( set[str], set[str], dict[str, tuple[bool, object, bool, object]], + bool, ] ] = [] if not full_delta: @@ -152,6 +153,7 @@ async def _patch_state( set(state.dirty_substates), computed_vars_to_preserve, computed_var_snapshots, + state._was_touched, )) states_to_snapshot.extend(state.substates.values()) root_state.dirty_vars.update(ROUTER_VARS) @@ -175,6 +177,7 @@ async def _patch_state( dirty_substates, computed_vars_to_preserve, computed_var_snapshots, + _, ) in dirty_state_snapshots: router_dirty_snapshots.append(( state, @@ -198,9 +201,11 @@ async def _patch_state( dirty_substates, _, computed_var_snapshots, + was_touched, ) in dirty_state_snapshots: state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates + state._was_touched = was_touched for name, computed_var in state.computed_vars.items(): ( had_cache, diff --git a/tests/units/test_state.py b/tests/units/test_state.py index c2c77ad681d..0edfdfdd7b5 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -1162,6 +1162,7 @@ async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): hasattr(linked_state, last_updated_attr), getattr(linked_state, last_updated_attr, None), ) + was_touched_before = linked_state._was_touched private_tree._clean() resolve_delta = private_tree._get_resolved_delta @@ -1179,6 +1180,7 @@ async def resolve_then_fail(): assert private_tree.substates[shared_state_name] is linked_state assert linked_state.dirty_vars == set() + assert linked_state._was_touched is was_touched_before assert ( hasattr(linked_state, cache_attr), getattr(linked_state, cache_attr, None), From 0b92c587a10517d25365b40af19b4db925ccc86d Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 15 Aug 2026 13:26:42 +0530 Subject: [PATCH 09/19] Handle non-scalar computed cache snapshots --- reflex/istate/shared.py | 13 ++++++++----- tests/units/test_state.py | 1 + 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 16160daae1e..2682083272c 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -5,7 +5,7 @@ import logging from collections.abc import AsyncIterator from dataclasses import MISSING -from typing import TYPE_CHECKING, TypeVar +from typing import Any, TYPE_CHECKING, TypeVar from reflex_base.constants import ROUTER_DATA, ROUTER_VARS from reflex_base.event import Event, get_hydrate_event @@ -125,7 +125,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, object, bool, object]], + dict[str, tuple[bool, Any, bool, Any]], bool, ] ] = [] @@ -167,7 +167,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, object, bool, object]], + dict[str, tuple[bool, Any, bool, Any]], ] ] = [] if not full_delta: @@ -246,10 +246,13 @@ async def _patch_state( computed_vars_refreshed.update( name for name, computed_var in state.computed_vars.items() - if computed_var_snapshots[name] + if ( + computed_var_snapshots[name][0], + computed_var_snapshots[name][2], + computed_var_snapshots[name][3], + ) != ( hasattr(state, computed_var._cache_attr), - getattr(state, computed_var._cache_attr, None), hasattr(state, computed_var._last_updated_attr), getattr(state, computed_var._last_updated_attr, None), ) diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 0edfdfdd7b5..43f5ce844d0 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -1169,6 +1169,7 @@ async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): async def resolve_then_fail(): linked_state._mark_dirty() + linked_state._was_touched = True await resolve_delta() msg = "computed refresh failed" raise RuntimeError(msg) From d4f64bc27c0edfadd8d0ae394fb98cf2bc9d8222 Mon Sep 17 00:00:00 2001 From: Harsh Thakare <151563948+harsh21234i@users.noreply.github.com> Date: Sun, 27 Sep 2026 16:59:11 +0530 Subject: [PATCH 10/19] style: sort shared state imports --- reflex/istate/shared.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 2682083272c..22268307589 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -5,7 +5,7 @@ import logging from collections.abc import AsyncIterator from dataclasses import MISSING -from typing import Any, TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING, Any, TypeVar from reflex_base.constants import ROUTER_DATA, ROUTER_VARS from reflex_base.event import Event, get_hydrate_event From 5f3661606215851e95d24b9917d0c23398e56523 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 2 Oct 2026 17:33:26 +0530 Subject: [PATCH 11/19] Fix shared-state computed delta filtering --- reflex/istate/shared.py | 38 ++-- tests/units/istate/test_shared.py | 299 ++++++++++++++++++++++++++++++ tests/units/test_state.py | 271 --------------------------- 3 files changed, 318 insertions(+), 290 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 22268307589..dcb9a5ae666 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -11,7 +11,7 @@ from reflex_base.event import Event, get_hydrate_event from reflex_base.registry import RegistrationContext from reflex_base.utils.exceptions import ReflexRuntimeError -from reflex_base.vars.base import _owner_state +from reflex_base.vars.base import _delta_value_key, _owner_state from typing_extensions import Self from reflex.istate.delta import _suppress_delta_recording @@ -125,7 +125,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, Any, bool, Any]], + dict[str, tuple[bool, Any, bool, Any, Any]], bool, ] ] = [] @@ -138,15 +138,17 @@ async def _patch_state( computed_vars_to_preserve = state._expired_computed_vars().union( state._always_dirty_computed_vars ) - computed_var_snapshots = { - name: ( - hasattr(state, computed_var._cache_attr), - getattr(state, computed_var._cache_attr, None), + computed_var_snapshots: dict[str, tuple[bool, Any, bool, Any, Any]] = {} + for name, computed_var in state.computed_vars.items(): + had_cache = hasattr(state, computed_var._cache_attr) + cached_value = getattr(state, computed_var._cache_attr, None) + computed_var_snapshots[name] = ( + had_cache, + cached_value, hasattr(state, computed_var._last_updated_attr), getattr(state, computed_var._last_updated_attr, None), + _delta_value_key(cached_value) if had_cache else None, ) - for name, computed_var in state.computed_vars.items() - } dirty_state_snapshots.append(( state, set(state.dirty_vars), @@ -167,7 +169,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, Any, bool, Any]], + dict[str, tuple[bool, Any, bool, Any, Any]], ] ] = [] if not full_delta: @@ -212,6 +214,7 @@ async def _patch_state( cached_value, had_last_updated, last_updated, + _, ) = computed_var_snapshots[name] if had_cache: setattr(state, computed_var._cache_attr, cached_value) @@ -246,17 +249,14 @@ async def _patch_state( computed_vars_refreshed.update( name for name, computed_var in state.computed_vars.items() - if ( - computed_var_snapshots[name][0], - computed_var_snapshots[name][2], - computed_var_snapshots[name][3], - ) - != ( - hasattr(state, computed_var._cache_attr), - hasattr(state, computed_var._last_updated_attr), - getattr(state, computed_var._last_updated_attr, None), + if hasattr(state, computed_var._cache_attr) + and ( + not computed_var_snapshots[name][0] + or computed_var_snapshots[name][4] + != _delta_value_key( + getattr(state, computed_var._cache_attr) + ) ) - and hasattr(state, computed_var._cache_attr) ) if computed_vars_refreshed: computed_refresh_states.append(state) diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index 4bcb4f8fe25..dbce9286e0e 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -6,6 +6,8 @@ from unittest.mock import AsyncMock, Mock, patch import pytest +from reflex_base import constants +from reflex_base.constants.state import FIELD_MARKER import reflex as rx from reflex.istate.shared import _do_update_other_tokens, _patch_state @@ -149,3 +151,300 @@ async def test_patch_state_recomputes_readers_after_restoring(): async with _patch_state(original_state=original, linked_state=linked): assert await reader.greeting == "linked" # pyright: ignore[reportAttributeAccessIssue] assert await reader.greeting == "private" # pyright: ignore[reportAttributeAccessIssue] + + +_PATCH_TEMPORARY_VAR = "temporary" +_PATCH_INTERVAL_VALUE_VAR = "interval_value" +_PATCH_ROOT_VALUE_VAR = "root_value" +_PATCH_ROUTER_VALUE_VAR = "router_client_token" +_PATCH_EXISTING_SUBSTATE = "existing" + + +class _LinkedStatePatchRoot(BaseState): + """Root state for testing linked-state dirty propagation.""" + + value: int = 0 + + +class _LinkedStatePatchShared(_LinkedStatePatchRoot): + """Substate used to exercise _patch_state without full SharedState setup.""" + + counter: int = 0 + + @rx.var + def root_value(self) -> int: + return self.value + + +class _LinkedStatePatchIntervalRoot(BaseState): + """Root state for testing interval computed refreshes during patching.""" + + value: int = 0 + + +class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): + """Substate used to exercise interval computed refreshes.""" + + counter: int = 0 + + @rx.var(interval=60) + def interval_value(self) -> int: + return self.value + + +class _LinkedStatePatchRouterRoot(BaseState): + """Root state with a computed value derived from the router.""" + + @rx.var + def router_client_token(self) -> str: + return self.router.session.client_token + + +class _LinkedStatePatchRouterShared(_LinkedStatePatchRouterRoot): + """Substate used to exercise root router-dependent computations.""" + + counter: int = 0 + + +@pytest.mark.asyncio +async def test_linked_state_event_does_not_dirty_root_state(): + """Linked-state events should not leak temporary router dirtiness.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + linked_state.counter = 1 + + assert set(constants.ROUTER_VARS).isdisjoint(private_tree.dirty_vars) + assert constants.ROUTER_DATA not in private_tree.dirty_vars + assert private_tree.get_full_name() not in private_tree.get_delta() + + +@pytest.mark.asyncio +async def test_linked_state_patch_does_not_emit_unchanged_router_computed_var(): + """Refreshing a router computed var should not emit it when unchanged.""" + private_tree = _LinkedStatePatchRouterRoot() + linked_tree = _LinkedStatePatchRouterRoot() + + shared_state_name = _LinkedStatePatchRouterShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + original_value = private_tree.router_client_token + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.router_client_token == original_value + assert _PATCH_ROUTER_VALUE_VAR not in private_tree.dirty_vars + assert private_tree.get_full_name() not in private_tree.get_delta() + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_root_dirty_state_on_resolve_error(): + """Temporary root dirtiness should be cleaned if delta resolution fails.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree.value = 1 + private_tree.dirty_substates.add(_PATCH_EXISTING_SUBSTATE) + original_dirty_vars = set(private_tree.dirty_vars) + original_dirty_substates = set(private_tree.dirty_substates) + + async def raise_resolve_error(): + await asyncio.sleep(0) + msg = "delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert private_tree.dirty_vars == original_dirty_vars + assert private_tree.dirty_substates == original_dirty_substates + assert private_tree.substates[shared_state_name] is private_state + assert linked_state.parent_state is linked_tree + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_on_resolve_error(): + """Temporary descendant dirtiness should be cleaned on resolution failure.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def raise_resolve_error(): + await asyncio.sleep(0) + linked_state.dirty_vars.add(_PATCH_TEMPORARY_VAR) + linked_state._mark_dirty() + msg = "descendant delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="descendant delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_after_resolve(): + """Temporary descendant dirtiness should be cleaned after resolution.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def resolve_with_temporary_dirty_state(): + await asyncio.sleep(0) + linked_state.dirty_vars.add(_PATCH_TEMPORARY_VAR) + linked_state._mark_dirty() + return {} + + object.__setattr__( + private_tree, "_get_resolved_delta", resolve_with_temporary_dirty_state + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_interval_computed_refresh(): + """Interval computed vars refreshed during patching must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_dirty_child(): + linked_state._mark_dirty() + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_dirty_child) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + interval_delta = private_tree.get_delta() + + assert _PATCH_INTERVAL_VALUE_VAR in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} + assert ( + _PATCH_INTERVAL_VALUE_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): + """Computed vars invalidated through the root must remain in the delta.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_root_change(): + private_tree.value = 1 + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_root_change) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + root_dependent_delta = private_tree.get_delta() + + assert _PATCH_ROOT_VALUE_VAR in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} + assert ( + _PATCH_ROOT_VALUE_VAR + FIELD_MARKER + in root_dependent_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): + """Failed resolution should not hide a partially refreshed computed value.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + computed_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_INTERVAL_VALUE_VAR + ] + cache_attr = computed_var._cache_attr + last_updated_attr = computed_var._last_updated_attr + cache_before = ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) + was_touched_before = linked_state._was_touched + + private_tree._clean() + resolve_delta = private_tree._get_resolved_delta + + async def resolve_then_fail(): + linked_state._mark_dirty() + linked_state._was_touched = True + await resolve_delta() + msg = "computed refresh failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_then_fail) + + with pytest.raises(RuntimeError, match="computed refresh failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + assert linked_state.dirty_vars == set() + assert linked_state._was_touched is was_touched_before + assert ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) == cache_before diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 43f5ce844d0..d0c1192779e 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -919,277 +919,6 @@ def test_setting_inherited_backend_var_does_not_mark_child_touched( assert not child_touched -class _LinkedStatePatchRoot(BaseState): - """Root state for testing linked-state dirty propagation.""" - - value: int = 0 - - -class _LinkedStatePatchShared(_LinkedStatePatchRoot): - """Substate used to exercise _patch_state without full SharedState setup.""" - - counter: int = 0 - - @rx.var - def root_value(self) -> int: - return self.value - - -class _LinkedStatePatchIntervalRoot(BaseState): - """Root state for testing interval computed refreshes during patching.""" - - value: int = 0 - - -class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): - """Substate used to exercise interval computed refreshes.""" - - counter: int = 0 - - @rx.var(interval=60) - def interval_value(self) -> int: - return self.value - - -@pytest.mark.asyncio -async def test_linked_state_event_does_not_dirty_root_state(): - """Linked-state events should not leak temporary router dirtiness.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchRoot() - linked_tree = _LinkedStatePatchRoot() - - shared_state_name = _LinkedStatePatchShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - assert isinstance(private_state, _LinkedStatePatchShared) - assert isinstance(linked_state, _LinkedStatePatchShared) - - private_tree._clean() - - async with _patch_state(private_state, linked_state, full_delta=False): - linked_state.counter = 1 - - assert "router" not in private_tree.dirty_vars - assert constants.ROUTER_DATA not in private_tree.dirty_vars - assert private_tree.get_full_name() not in private_tree.get_delta() - - -@pytest.mark.asyncio -async def test_linked_state_patch_restores_root_dirty_state_on_resolve_error( - monkeypatch: pytest.MonkeyPatch, -): - """Temporary root dirtiness should be cleaned if delta resolution fails.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchRoot() - linked_tree = _LinkedStatePatchRoot() - - shared_state_name = _LinkedStatePatchShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - assert isinstance(private_state, _LinkedStatePatchShared) - assert isinstance(linked_state, _LinkedStatePatchShared) - - private_tree.value = 1 - private_tree.dirty_substates.add("existing") - original_dirty_vars = set(private_tree.dirty_vars) - original_dirty_substates = set(private_tree.dirty_substates) - - async def raise_resolve_error(): - await asyncio.sleep(0) - msg = "delta resolution failed" - raise RuntimeError(msg) - - object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) - - with pytest.raises(RuntimeError, match="delta resolution failed"): - async with _patch_state(private_state, linked_state, full_delta=False): - pass - - assert private_tree.dirty_vars == original_dirty_vars - assert private_tree.dirty_substates == original_dirty_substates - assert private_tree.substates[shared_state_name] is private_state - assert linked_state.parent_state is linked_tree - - -@pytest.mark.asyncio -async def test_linked_state_patch_restores_descendant_dirty_state_on_resolve_error(): - """Temporary descendant dirtiness should be cleaned on resolution failure.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchRoot() - linked_tree = _LinkedStatePatchRoot() - - shared_state_name = _LinkedStatePatchShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - async def raise_resolve_error(): - await asyncio.sleep(0) - linked_state.dirty_vars.add("temporary") - linked_state._mark_dirty() - msg = "descendant delta resolution failed" - raise RuntimeError(msg) - - object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) - - with pytest.raises(RuntimeError, match="descendant delta resolution failed"): - async with _patch_state(private_state, linked_state, full_delta=False): - pass - - assert linked_state.dirty_vars == set() - assert linked_state.dirty_substates == set() - assert private_tree.dirty_substates == set() - - -@pytest.mark.asyncio -async def test_linked_state_patch_restores_descendant_dirty_state_after_resolve(): - """Temporary descendant dirtiness should be cleaned after resolution.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchRoot() - linked_tree = _LinkedStatePatchRoot() - - shared_state_name = _LinkedStatePatchShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - async def resolve_with_temporary_dirty_state(): - await asyncio.sleep(0) - linked_state.dirty_vars.add("temporary") - linked_state._mark_dirty() - return {} - - object.__setattr__( - private_tree, "_get_resolved_delta", resolve_with_temporary_dirty_state - ) - - async with _patch_state(private_state, linked_state, full_delta=False): - assert private_tree.substates[shared_state_name] is linked_state - - assert linked_state.dirty_vars == set() - assert linked_state.dirty_substates == set() - assert private_tree.dirty_substates == set() - - -@pytest.mark.asyncio -async def test_linked_state_patch_preserves_interval_computed_refresh(): - """Interval computed vars refreshed during patching must be emitted.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchIntervalRoot() - linked_tree = _LinkedStatePatchIntervalRoot() - - shared_state_name = _LinkedStatePatchIntervalShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - private_tree._clean() - - resolve_delta = private_tree._get_resolved_delta - - async def resolve_with_dirty_child(): - linked_state._mark_dirty() - return await resolve_delta() - - object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_dirty_child) - - async with _patch_state(private_state, linked_state, full_delta=False): - assert private_tree.substates[shared_state_name] is linked_state - interval_delta = private_tree.get_delta() - - assert "interval_value" in linked_state.dirty_vars - assert private_tree.dirty_substates == {linked_state.get_name()} - assert ( - "interval_value" + FIELD_MARKER in interval_delta[linked_state.get_full_name()] - ) - - -@pytest.mark.asyncio -async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): - """Computed vars invalidated through the root must remain in the delta.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchRoot() - linked_tree = _LinkedStatePatchRoot() - - shared_state_name = _LinkedStatePatchShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - - private_tree._clean() - - resolve_delta = private_tree._get_resolved_delta - - async def resolve_with_root_change(): - private_tree.value = 1 - return await resolve_delta() - - object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_root_change) - - async with _patch_state(private_state, linked_state, full_delta=False): - assert private_tree.substates[shared_state_name] is linked_state - root_dependent_delta = private_tree.get_delta() - - assert "root_value" in linked_state.dirty_vars - assert private_tree.dirty_substates == {linked_state.get_name()} - assert ( - "root_value" + FIELD_MARKER - in root_dependent_delta[linked_state.get_full_name()] - ) - - -@pytest.mark.asyncio -async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): - """Failed resolution should not hide a partially refreshed computed value.""" - from reflex.istate.shared import _patch_state - - private_tree = _LinkedStatePatchIntervalRoot() - linked_tree = _LinkedStatePatchIntervalRoot() - - shared_state_name = _LinkedStatePatchIntervalShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] - computed_var = _LinkedStatePatchIntervalShared.computed_vars["interval_value"] - cache_attr = computed_var._cache_attr - last_updated_attr = computed_var._last_updated_attr - cache_before = ( - hasattr(linked_state, cache_attr), - getattr(linked_state, cache_attr, None), - hasattr(linked_state, last_updated_attr), - getattr(linked_state, last_updated_attr, None), - ) - was_touched_before = linked_state._was_touched - - private_tree._clean() - resolve_delta = private_tree._get_resolved_delta - - async def resolve_then_fail(): - linked_state._mark_dirty() - linked_state._was_touched = True - await resolve_delta() - msg = "computed refresh failed" - raise RuntimeError(msg) - - object.__setattr__(private_tree, "_get_resolved_delta", resolve_then_fail) - - with pytest.raises(RuntimeError, match="computed refresh failed"): - async with _patch_state(private_state, linked_state, full_delta=False): - assert private_tree.substates[shared_state_name] is linked_state - - assert linked_state.dirty_vars == set() - assert linked_state._was_touched is was_touched_before - assert ( - hasattr(linked_state, cache_attr), - getattr(linked_state, cache_attr, None), - hasattr(linked_state, last_updated_attr), - getattr(linked_state, last_updated_attr, None), - ) == cache_before - - @pytest.mark.asyncio async def test_process_event_simple( token: str, From b2fe24713f87b65036a71684a430a87b0f933c48 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 2 Oct 2026 17:53:29 +0530 Subject: [PATCH 12/19] Limit linked-state cache comparisons to router deps --- reflex/istate/shared.py | 39 ++++++++++++++++++++----------- tests/units/istate/test_shared.py | 35 +++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 13 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index dcb9a5ae666..dea1ac3dcc8 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -125,7 +125,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, Any, bool, Any, Any]], + dict[str, tuple[bool, Any, bool, Any]], bool, ] ] = [] @@ -138,7 +138,7 @@ async def _patch_state( computed_vars_to_preserve = state._expired_computed_vars().union( state._always_dirty_computed_vars ) - computed_var_snapshots: dict[str, tuple[bool, Any, bool, Any, Any]] = {} + computed_var_snapshots: dict[str, tuple[bool, Any, bool, Any]] = {} for name, computed_var in state.computed_vars.items(): had_cache = hasattr(state, computed_var._cache_attr) cached_value = getattr(state, computed_var._cache_attr, None) @@ -147,7 +147,6 @@ async def _patch_state( cached_value, hasattr(state, computed_var._last_updated_attr), getattr(state, computed_var._last_updated_attr, None), - _delta_value_key(cached_value) if had_cache else None, ) dirty_state_snapshots.append(( state, @@ -169,7 +168,7 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, Any, bool, Any, Any]], + dict[str, tuple[bool, Any]], ] ] = [] if not full_delta: @@ -181,14 +180,26 @@ async def _patch_state( computed_var_snapshots, _, ) in dirty_state_snapshots: + router_dirty_vars = ( + state.dirty_vars - dirty_vars - computed_vars_to_preserve + ) + router_computed_snapshots = { + name: ( + computed_var_snapshots[name][0], + _delta_value_key(computed_var_snapshots[name][1]) + if computed_var_snapshots[name][0] + else None, + ) + for name in router_dirty_vars.intersection(state.computed_vars) + } router_dirty_snapshots.append(( state, dirty_vars, dirty_substates, - state.dirty_vars - dirty_vars - computed_vars_to_preserve, + router_dirty_vars, state.dirty_substates - dirty_substates, computed_vars_to_preserve, - computed_var_snapshots, + router_computed_snapshots, )) try: # The delta is discarded: it is only resolved to refresh computed vars, @@ -214,7 +225,6 @@ async def _patch_state( cached_value, had_last_updated, last_updated, - _, ) = computed_var_snapshots[name] if had_cache: setattr(state, computed_var._cache_attr, cached_value) @@ -241,20 +251,23 @@ async def _patch_state( router_dirty_vars, _, computed_vars_to_preserve, - computed_var_snapshots, + router_computed_snapshots, ) in router_dirty_snapshots: computed_vars_refreshed = set(computed_vars_to_preserve) | ( state.dirty_vars - router_dirty_vars ).intersection(state.computed_vars) computed_vars_refreshed.update( name - for name, computed_var in state.computed_vars.items() - if hasattr(state, computed_var._cache_attr) + for name, ( + had_cache, + cached_value_key, + ) in router_computed_snapshots.items() + if hasattr(state, state.computed_vars[name]._cache_attr) and ( - not computed_var_snapshots[name][0] - or computed_var_snapshots[name][4] + not had_cache + or cached_value_key != _delta_value_key( - getattr(state, computed_var._cache_attr) + getattr(state, state.computed_vars[name]._cache_attr) ) ) ) diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index dbce9286e0e..8cab12e51e0 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -199,6 +199,10 @@ class _LinkedStatePatchRouterRoot(BaseState): def router_client_token(self) -> str: return self.router.session.client_token + @rx.var + def unrelated_values(self) -> list[int]: + return list(range(1000)) + class _LinkedStatePatchRouterShared(_LinkedStatePatchRouterRoot): """Substate used to exercise root router-dependent computations.""" @@ -248,6 +252,37 @@ async def test_linked_state_patch_does_not_emit_unchanged_router_computed_var(): assert private_tree.get_full_name() not in private_tree.get_delta() +@pytest.mark.asyncio +async def test_linked_state_patch_skips_unrelated_computed_value_keys( + monkeypatch: pytest.MonkeyPatch, +): + """Only router-dependent computed values need a cache comparison key.""" + from reflex.istate import shared as shared_module + + private_tree = _LinkedStatePatchRouterRoot() + linked_tree = _LinkedStatePatchRouterRoot() + shared_state_name = _LinkedStatePatchRouterShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + unrelated_values = private_tree.unrelated_values + private_tree._clean() + delta_value_key = shared_module._delta_value_key + unrelated_key_calls = [] + + def track_unrelated_value_key(value): + if value is unrelated_values: + unrelated_key_calls.append(value) + return delta_value_key(value) + + monkeypatch.setattr(shared_module, "_delta_value_key", track_unrelated_value_key) + + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert unrelated_key_calls == [] + + @pytest.mark.asyncio async def test_linked_state_patch_restores_root_dirty_state_on_resolve_error(): """Temporary root dirtiness should be cleaned if delta resolution fails.""" From 72d4db5e2aa3c29db05484206dd316d0b3e51815 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 2 Oct 2026 18:23:57 +0530 Subject: [PATCH 13/19] Preserve interval refreshes in linked state deltas --- reflex/istate/shared.py | 125 ++++++++++++++++++++++++------ tests/units/istate/test_shared.py | 41 ++++++++++ 2 files changed, 144 insertions(+), 22 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index dea1ac3dcc8..ceb741e64b2 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -125,10 +125,10 @@ async def _patch_state( set[str], set[str], set[str], - dict[str, tuple[bool, Any, bool, Any]], bool, ] ] = [] + states_by_name: dict[str, BaseState] = {} if not full_delta: states_to_snapshot = [root_state] while states_to_snapshot: @@ -138,25 +138,82 @@ async def _patch_state( computed_vars_to_preserve = state._expired_computed_vars().union( state._always_dirty_computed_vars ) - computed_var_snapshots: dict[str, tuple[bool, Any, bool, Any]] = {} - for name, computed_var in state.computed_vars.items(): - had_cache = hasattr(state, computed_var._cache_attr) - cached_value = getattr(state, computed_var._cache_attr, None) - computed_var_snapshots[name] = ( - had_cache, - cached_value, - hasattr(state, computed_var._last_updated_attr), - getattr(state, computed_var._last_updated_attr, None), - ) + states_by_name[state.get_full_name()] = state dirty_state_snapshots.append(( state, set(state.dirty_vars), set(state.dirty_substates), computed_vars_to_preserve, - computed_var_snapshots, state._was_touched, )) states_to_snapshot.extend(state.substates.values()) + computed_vars_to_snapshot = { + state.get_full_name(): ( + dirty_vars.intersection(state.computed_vars) + | computed_vars_to_preserve + | state._interval_computed_var_names + ) + for ( + state, + dirty_vars, + _, + computed_vars_to_preserve, + _, + ) in dirty_state_snapshots + } + # Track only caches that dirty propagation can invalidate, before + # marking router vars deletes those cached values. + pending_dependencies = [ + (state, name) + for ( + state, + dirty_vars, + _, + computed_vars_to_preserve, + _, + ) in dirty_state_snapshots + for name in dirty_vars | computed_vars_to_preserve + ] + pending_dependencies.extend((root_state, name) for name in ROUTER_VARS) + pending_dependencies.append((root_state, ROUTER_DATA)) + seen_dependencies: set[tuple[str, str]] = set() + while pending_dependencies: + state, name = pending_dependencies.pop() + dependency = (state.get_full_name(), name) + if dependency in seen_dependencies: + continue + seen_dependencies.add(dependency) + for target_name, computed_var_name in state._var_dependencies.get( + name, () + ): + target_state = states_by_name.get(target_name) + if target_state is None: + continue + target_vars = computed_vars_to_snapshot[target_name] + if computed_var_name not in target_vars: + target_vars.add(computed_var_name) + pending_dependencies.append((target_state, computed_var_name)) + computed_var_cache_snapshots: dict[ + str, dict[str, tuple[bool, Any, bool, Any]] + ] = {} + for ( + state, + _, + _, + _, + _, + ) in dirty_state_snapshots: + computed_var_cache_snapshots[state.get_full_name()] = { + name: ( + hasattr(state, state.computed_vars[name]._cache_attr), + getattr(state, state.computed_vars[name]._cache_attr, None), + hasattr(state, state.computed_vars[name]._last_updated_attr), + getattr( + state, state.computed_vars[name]._last_updated_attr, None + ), + ) + for name in computed_vars_to_snapshot[state.get_full_name()] + } root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() @@ -169,6 +226,7 @@ async def _patch_state( set[str], set[str], dict[str, tuple[bool, Any]], + dict[str, tuple[bool, Any, bool, Any]], ] ] = [] if not full_delta: @@ -177,9 +235,11 @@ async def _patch_state( dirty_vars, dirty_substates, computed_vars_to_preserve, - computed_var_snapshots, _, ) in dirty_state_snapshots: + computed_var_snapshots = computed_var_cache_snapshots[ + state.get_full_name() + ] router_dirty_vars = ( state.dirty_vars - dirty_vars - computed_vars_to_preserve ) @@ -190,7 +250,7 @@ async def _patch_state( if computed_var_snapshots[name][0] else None, ) - for name in router_dirty_vars.intersection(state.computed_vars) + for name in router_dirty_vars.intersection(computed_var_snapshots) } router_dirty_snapshots.append(( state, @@ -200,6 +260,7 @@ async def _patch_state( state.dirty_substates - dirty_substates, computed_vars_to_preserve, router_computed_snapshots, + computed_var_snapshots, )) try: # The delta is discarded: it is only resolved to refresh computed vars, @@ -213,19 +274,21 @@ async def _patch_state( dirty_vars, dirty_substates, _, - computed_var_snapshots, was_touched, ) in dirty_state_snapshots: + computed_var_snapshots = computed_var_cache_snapshots[ + state.get_full_name() + ] state.dirty_vars = dirty_vars state.dirty_substates = dirty_substates state._was_touched = was_touched - for name, computed_var in state.computed_vars.items(): - ( - had_cache, - cached_value, - had_last_updated, - last_updated, - ) = computed_var_snapshots[name] + for name, ( + had_cache, + cached_value, + had_last_updated, + last_updated, + ) in computed_var_snapshots.items(): + computed_var = state.computed_vars[name] if had_cache: setattr(state, computed_var._cache_attr, cached_value) else: @@ -252,10 +315,28 @@ async def _patch_state( _, computed_vars_to_preserve, router_computed_snapshots, + computed_var_snapshots, ) in router_dirty_snapshots: computed_vars_refreshed = set(computed_vars_to_preserve) | ( state.dirty_vars - router_dirty_vars ).intersection(state.computed_vars) + computed_vars_refreshed.update( + name + for name in state._interval_computed_var_names + if ( + computed_var_snapshots[name][2] + != hasattr( + state, + state.computed_vars[name]._last_updated_attr, + ) + or computed_var_snapshots[name][3] + != getattr( + state, + state.computed_vars[name]._last_updated_attr, + None, + ) + ) + ) computed_vars_refreshed.update( name for name, ( diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index 8cab12e51e0..65c7b7bb203 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -1,6 +1,7 @@ """Unit tests for shared state fan-out to other linked clients.""" import asyncio +import datetime import pickle from contextlib import asynccontextmanager from unittest.mock import AsyncMock, Mock, patch @@ -406,6 +407,46 @@ async def resolve_with_dirty_child(): ) +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_interval_refresh_during_async_router_var(): + """An interval var refreshed during async resolution must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert linked_state.interval_value == linked_state.value + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_after_async_router_computation(): + await asyncio.sleep(0) + object.__setattr__(linked_state, "value", 1) + interval_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_INTERVAL_VALUE_VAR + ] + setattr(linked_state, interval_var._last_updated_attr, datetime.datetime.min) + assert linked_state.interval_value == 1 + return await resolve_delta() + + object.__setattr__( + private_tree, + "_get_resolved_delta", + resolve_after_async_router_computation, + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + interval_delta = private_tree.get_delta() + + assert _PATCH_INTERVAL_VALUE_VAR in linked_state.dirty_vars + assert ( + _PATCH_INTERVAL_VALUE_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + @pytest.mark.asyncio async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): """Computed vars invalidated through the root must remain in the delta.""" From 50f8bd65fc750921609c8ea2d4790f933f0f5e18 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 2 Oct 2026 18:41:24 +0530 Subject: [PATCH 14/19] Preserve downstream interval computed updates --- reflex/istate/shared.py | 11 +++--- tests/units/istate/test_shared.py | 66 ++++++++++++++++++++++++++++++- 2 files changed, 69 insertions(+), 8 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index ceb741e64b2..86d67574148 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -129,6 +129,9 @@ async def _patch_state( ] ] = [] states_by_name: dict[str, BaseState] = {} + computed_var_cache_snapshots: dict[ + str, dict[str, tuple[bool, Any, bool, Any]] + ] = {} if not full_delta: states_to_snapshot = [root_state] while states_to_snapshot: @@ -190,12 +193,8 @@ async def _patch_state( if target_state is None: continue target_vars = computed_vars_to_snapshot[target_name] - if computed_var_name not in target_vars: - target_vars.add(computed_var_name) - pending_dependencies.append((target_state, computed_var_name)) - computed_var_cache_snapshots: dict[ - str, dict[str, tuple[bool, Any, bool, Any]] - ] = {} + target_vars.add(computed_var_name) + pending_dependencies.append((target_state, computed_var_name)) for ( state, _, diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index 65c7b7bb203..4e631f9d530 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -4,6 +4,7 @@ import datetime import pickle from contextlib import asynccontextmanager +from typing import cast from unittest.mock import AsyncMock, Mock, patch import pytest @@ -11,6 +12,7 @@ from reflex_base.constants.state import FIELD_MARKER import reflex as rx +from reflex.istate.data import RouterData, SessionData from reflex.istate.shared import _do_update_other_tokens, _patch_state from reflex.state import BaseState, State from reflex.utils.token_manager import ( @@ -156,6 +158,8 @@ async def test_patch_state_recomputes_readers_after_restoring(): _PATCH_TEMPORARY_VAR = "temporary" _PATCH_INTERVAL_VALUE_VAR = "interval_value" +_PATCH_ROUTER_INTERVAL_VALUE_VAR = "router_interval_value" +_PATCH_ROUTER_INTERVAL_DEPENDENT_VAR = "router_interval_dependent_value" _PATCH_ROOT_VALUE_VAR = "root_value" _PATCH_ROUTER_VALUE_VAR = "router_client_token" _PATCH_EXISTING_SUBSTATE = "existing" @@ -192,6 +196,14 @@ class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): def interval_value(self) -> int: return self.value + @rx.var(interval=60) + def router_interval_value(self) -> str: + return self.router.session.client_token + + @rx.var + def router_interval_dependent_value(self) -> str: + return self.router_interval_value + "-dependent" + class _LinkedStatePatchRouterRoot(BaseState): """Root state with a computed value derived from the router.""" @@ -413,8 +425,12 @@ async def test_linked_state_patch_preserves_interval_refresh_during_async_router private_tree = _LinkedStatePatchIntervalRoot() linked_tree = _LinkedStatePatchIntervalRoot() shared_state_name = _LinkedStatePatchIntervalShared.get_name() - private_state = private_tree.substates[shared_state_name] - linked_state = linked_tree.substates[shared_state_name] + private_state = cast( + _LinkedStatePatchIntervalShared, private_tree.substates[shared_state_name] + ) + linked_state = cast( + _LinkedStatePatchIntervalShared, linked_tree.substates[shared_state_name] + ) assert linked_state.interval_value == linked_state.value private_tree._clean() @@ -447,6 +463,52 @@ async def resolve_after_async_router_computation(): ) +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_downstream_of_refreshed_interval_var(): + """A downstream computed var of a refreshed interval var must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + private_tree.router = RouterData(session=SessionData(client_token="before")) + linked_tree.router = RouterData(session=SessionData(client_token="before")) + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = cast( + _LinkedStatePatchIntervalShared, private_tree.substates[shared_state_name] + ) + linked_state = cast( + _LinkedStatePatchIntervalShared, linked_tree.substates[shared_state_name] + ) + + assert linked_state.router_interval_dependent_value == "before-dependent" + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_after_router_computation(): + await asyncio.sleep(0) + private_tree.router = RouterData(session=SessionData(client_token="after")) + interval_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_ROUTER_INTERVAL_VALUE_VAR + ] + setattr(linked_state, interval_var._last_updated_attr, datetime.datetime.min) + assert linked_state.router_interval_dependent_value == "after-dependent" + return await resolve_delta() + + object.__setattr__( + private_tree, + "_get_resolved_delta", + resolve_after_router_computation, + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + interval_delta = private_tree.get_delta() + + assert _PATCH_ROUTER_INTERVAL_DEPENDENT_VAR in linked_state.dirty_vars + assert ( + _PATCH_ROUTER_INTERVAL_DEPENDENT_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + @pytest.mark.asyncio async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): """Computed vars invalidated through the root must remain in the delta.""" From e46fa8a714c9160dca82c6113fc32e71c78104bb Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 3 Oct 2026 12:05:47 +0530 Subject: [PATCH 15/19] test: make app name validation portable --- tests/units/utils/test_utils.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/units/utils/test_utils.py b/tests/units/utils/test_utils.py index 4ff41dac4cf..7aacad8027c 100644 --- a/tests/units/utils/test_utils.py +++ b/tests/units/utils/test_utils.py @@ -646,17 +646,17 @@ def test_initialize_requirements_txt_preserves_existing_requirements(tmp_path): assert not pyproject_file.exists() -def test_validate_app_name(tmp_path, mocker: MockerFixture): +def test_validate_app_name(tmp_path, monkeypatch: pytest.MonkeyPatch): """Test that an error is raised if the app name is reflex or if the name is not according to python package naming conventions. Args: tmp_path: Test working dir. - mocker: Pytest mocker object. + monkeypatch: Pytest monkeypatch fixture. """ reflex = tmp_path / "reflex" reflex.mkdir() - mocker.patch("os.getcwd", return_value=str(reflex)) + monkeypatch.chdir(reflex) with pytest.raises(SystemExit): prerequisites.validate_app_name() From bc8a6950c29e76728a5762e7886884c5f488665b Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 7 Oct 2026 15:14:35 +0530 Subject: [PATCH 16/19] Skip unnecessary linked-state router refresh --- reflex/istate/shared.py | 27 +++++++++++++++++++++++++++ tests/units/istate/test_shared.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index c560ed36692..d85f1d0221c 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -118,6 +118,33 @@ async def _patch_state( # it forces router-dependent computed vars to resolve for the patched # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() + if ( + not full_delta + and not any( + root_state._var_dependencies.get(name) + for name in (*ROUTER_VARS, ROUTER_DATA) + ) + and all( + name in ROUTER_VARS or name == ROUTER_DATA + for name in root_state.dirty_vars + ) + and not root_state._potentially_dirty_states + ): + states_to_check = [root_state] + while states_to_check: + state = states_to_check.pop() + if state.computed_vars: + break + states_to_check.extend( + state.substates[name] + for name in state.dirty_substates | state._always_dirty_substates + if name in state.substates + ) + else: + # There is no router-dependent or otherwise pending computed + # work, so resolving this discarded delta cannot change state. + yield + return dirty_state_snapshots: list[ tuple[ BaseState, diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index 2ac49f7bdbe..3cac930d47f 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -247,6 +247,16 @@ class _LinkedStatePatchRouterShared(_LinkedStatePatchRouterRoot): counter: int = 0 +class _LinkedStatePatchNoComputedRoot(BaseState): + """Root state with no computed vars for the discarded-delta fast path.""" + + +class _LinkedStatePatchNoComputedShared(_LinkedStatePatchNoComputedRoot): + """Substate with no computed vars for the discarded-delta fast path.""" + + counter: int = 0 + + @pytest.mark.asyncio async def test_linked_state_event_does_not_dirty_root_state(): """Linked-state events should not leak temporary router dirtiness.""" @@ -270,6 +280,26 @@ async def test_linked_state_event_does_not_dirty_root_state(): assert private_tree.get_full_name() not in private_tree.get_delta() +@pytest.mark.asyncio +async def test_linked_state_patch_skips_discarded_delta_without_computed_vars(): + """Skip router refresh when no computed var or pending substate needs it.""" + private_tree = _LinkedStatePatchNoComputedRoot() + linked_tree = _LinkedStatePatchNoComputedRoot() + shared_state_name = _LinkedStatePatchNoComputedShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + resolve_delta = AsyncMock() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_delta) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + resolve_delta.assert_not_awaited() + assert not private_tree.dirty_vars + assert not private_tree.dirty_substates + + @pytest.mark.asyncio async def test_linked_state_patch_does_not_emit_unchanged_router_computed_var(): """Refreshing a router computed var should not emit it when unchanged.""" From cca047032b5b3735177ec4c0db7c3bd2dc7aab48 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sat, 17 Oct 2026 19:09:46 +0530 Subject: [PATCH 17/19] test: move shared module import to module scope --- tests/units/istate/test_shared.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index 3cac930d47f..f5d127eace7 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -12,6 +12,7 @@ from reflex_base.constants.state import FIELD_MARKER import reflex as rx +from reflex.istate import shared as shared_module from reflex.istate.data import RouterData, SessionData from reflex.istate.shared import ( SharedStateBaseInternal, @@ -324,8 +325,6 @@ async def test_linked_state_patch_skips_unrelated_computed_value_keys( monkeypatch: pytest.MonkeyPatch, ): """Only router-dependent computed values need a cache comparison key.""" - from reflex.istate import shared as shared_module - private_tree = _LinkedStatePatchRouterRoot() linked_tree = _LinkedStatePatchRouterRoot() shared_state_name = _LinkedStatePatchRouterShared.get_name() From ae45ca2e08f6020060d49e75e895c77a17618adb Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 22 Oct 2026 16:44:08 +0530 Subject: [PATCH 18/19] test: synchronize redis lock contention queue --- tests/units/istate/manager/test_redis.py | 35 ++++++++++++++++++++---- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/tests/units/istate/manager/test_redis.py b/tests/units/istate/manager/test_redis.py index 5ed218e1ef4..96b616fe2bf 100644 --- a/tests/units/istate/manager/test_redis.py +++ b/tests/units/istate/manager/test_redis.py @@ -7,7 +7,7 @@ import time import uuid from collections import Counter -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, AsyncIterator from types import ModuleType from typing import Any from unittest.mock import AsyncMock, Mock @@ -424,6 +424,7 @@ async def test_oplock_contention_queue( state_manager_redis: StateManagerRedis, root_state: type[RedisTestState], event_log: list[dict[str, Any]], + monkeypatch: pytest.MonkeyPatch, ): """Test the oplock contention queue. @@ -431,6 +432,7 @@ async def test_oplock_contention_queue( state_manager_redis: The StateManagerRedis to test. root_state: The root state class. event_log: The redis event log. + monkeypatch: The pytest monkeypatch fixture. """ token = str(uuid.uuid4()) @@ -445,7 +447,6 @@ async def test_oplock_contention_queue( state_manager_2._oplock_enabled = True modify_started = asyncio.Event() - modify_2_started = asyncio.Event() modify_1_continue = asyncio.Event() modify_2_continue = asyncio.Event() @@ -460,7 +461,6 @@ async def modify_1(): async def modify_2(): await modify_started.wait() - modify_2_started.set() async with state_manager_2.modify_state( BaseStateToken(ident=token, cls=root_state), ) as new_state: @@ -470,7 +470,6 @@ async def modify_2(): async def modify_3(): await modify_started.wait() - modify_2_started.set() async with state_manager_2.modify_state( BaseStateToken(ident=token, cls=root_state), ) as new_state: @@ -481,8 +480,34 @@ async def modify_3(): task_1 = asyncio.create_task(modify_1()) task_2 = asyncio.create_task(modify_2()) task_3 = asyncio.create_task(modify_3()) + lock_key = state_manager_2._lock_key(BaseStateToken(ident=token, cls=root_state)) + queued_waiters = 0 + both_waiters_queued = asyncio.Event() + original_lock_waiter = StateManagerRedis._lock_waiter - await modify_2_started.wait() + @contextlib.asynccontextmanager + async def track_lock_waiter( + manager: StateManagerRedis, key: bytes + ) -> AsyncIterator[asyncio.Event]: + nonlocal queued_waiters + async with original_lock_waiter(manager, key) as lock_released_event: + if manager is state_manager_2 and key == lock_key: + queued_waiters += 1 + if queued_waiters == 2: + both_waiters_queued.set() + yield lock_released_event + + monkeypatch.setattr(StateManagerRedis, "_lock_waiter", track_lock_waiter) + + # Do not release the lock until both modifications have joined its queue. + try: + await asyncio.wait_for(both_waiters_queued.wait(), timeout=5) + except TimeoutError: + for task in (task_1, task_2, task_3): + task.cancel() + await asyncio.gather(task_1, task_2, task_3, return_exceptions=True) + await state_manager_2.close() + raise # Let modify 1 complete modify_1_continue.set() From 333562e25aa4a3512534756bf052c3750835f7cd Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Sun, 1 Nov 2026 12:03:03 +0530 Subject: [PATCH 19/19] fix: preserve synchronized storage updates in linked deltas --- reflex/istate/shared.py | 40 +++++++++++++++++++++------- reflex/state.py | 7 +++++ tests/units/istate/test_shared.py | 44 +++++++++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 10 deletions(-) diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index d85f1d0221c..2eb9d496ef2 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -15,7 +15,13 @@ from reflex.istate.delta import _suppress_delta_recording from reflex.istate.manager.token import BaseStateToken -from reflex.state import BaseState, OnLoadInternalState, State, _override_base_method +from reflex.state import ( + BaseState, + OnLoadInternalState, + State, + _override_base_method, + _shared_storage_updates, +) logger = logging.getLogger(__name__) @@ -287,11 +293,16 @@ async def _patch_state( router_computed_snapshots, computed_var_snapshots, )) + storage_updates: set[tuple[str, str]] = set() + storage_updates_token = _shared_storage_updates.set(storage_updates) try: - # The delta is discarded: it is only resolved to refresh computed vars, - # so its values must not count as sent to the client. - with _suppress_delta_recording(): - await root_state._get_resolved_delta() + try: + # The delta is discarded: it is only resolved to refresh computed vars, + # so its values must not count as sent to the client. + with _suppress_delta_recording(): + await root_state._get_resolved_delta() + finally: + _shared_storage_updates.reset(storage_updates_token) except BaseException: if not full_delta: for ( @@ -331,7 +342,7 @@ async def _patch_state( raise else: if not full_delta: - computed_refresh_states: list[BaseState] = [] + refreshed_states: list[BaseState] = [] for ( state, dirty_vars, @@ -377,13 +388,22 @@ async def _patch_state( ) ) ) - if computed_vars_refreshed: - computed_refresh_states.append(state) + field_vars_refreshed = { + var_name + for state_name, var_name in storage_updates + if state_name == state.get_full_name() + and var_name not in state.computed_vars + } + if computed_vars_refreshed or field_vars_refreshed: + refreshed_states.append(state) state.dirty_vars = ( - dirty_vars | computed_vars_to_preserve | computed_vars_refreshed + dirty_vars + | computed_vars_to_preserve + | computed_vars_refreshed + | field_vars_refreshed ) state.dirty_substates = dirty_substates - for refreshed_state in computed_refresh_states: + for refreshed_state in refreshed_states: state = refreshed_state while state.parent_state is not None: state.parent_state.dirty_substates.add(state.get_name()) diff --git a/reflex/state.py b/reflex/state.py index b4e20f2257b..c0711e10160 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -15,6 +15,7 @@ import re import sys from collections.abc import Callable, Iterable, Iterator, Sequence +from contextvars import ContextVar from datetime import timedelta from hashlib import md5 from types import FunctionType @@ -101,6 +102,10 @@ from reflex.utils import console, format, types from reflex.utils.exec import is_testing_env +_shared_storage_updates: ContextVar[set[tuple[str, str]] | None] = ContextVar( + "shared_storage_updates", default=None +) + @functools.cache def _stale_pickle_keys(cls: type[BaseState]) -> frozenset[str]: @@ -2353,6 +2358,8 @@ async def _sync_shared_storage(state: BaseState) -> None: target = await state.get_state(state_cls) if target.get_value(var_name) != value: setattr(target, var_name, value) + if (updates := _shared_storage_updates.get()) is not None: + updates.add((target.get_full_name(), var_name)) def _loaded_substate(root: BaseState, state_name: str) -> BaseState | None: diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index f5d127eace7..b8d96fb042b 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -10,6 +10,7 @@ import pytest from reflex_base import constants from reflex_base.constants.state import FIELD_MARKER +from reflex_base.registry import RegistrationContext import reflex as rx from reflex.istate import shared as shared_module @@ -440,6 +441,49 @@ async def resolve_with_temporary_dirty_state(): assert private_tree.dirty_substates == set() +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_synced_field_updates( + clean_registration_context: RegistrationContext, +): + """Fields synchronized during refresh must remain dirty for the final delta.""" + + class _StoragePatchRoot(BaseState): + @rx.var + def router_value(self) -> str: + return self.router.session.client_token + + class _StoragePatchSource(_StoragePatchRoot): + value: str = rx.LocalStorage("", name="shared", sync=True) + + class _StoragePatchSibling(_StoragePatchRoot): + value: str = rx.LocalStorage("", name="shared", sync=True) + + private_tree = _StoragePatchRoot() + linked_tree = _StoragePatchRoot() + private_state = cast( + _StoragePatchSource, + private_tree.substates[_StoragePatchSource.get_name()], + ) + linked_state = cast( + _StoragePatchSource, + linked_tree.substates[_StoragePatchSource.get_name()], + ) + sibling_state = cast( + _StoragePatchSibling, + private_tree.substates[_StoragePatchSibling.get_name()], + ) + private_tree._clean() + linked_state.value = "updated" + + async with _patch_state(private_state, linked_state, full_delta=False): + delta = private_tree.get_delta() + + assert sibling_state.value == "updated" + assert sibling_state.dirty_vars == {"value"} + assert sibling_state.get_name() in private_tree.dirty_substates + assert delta[sibling_state.get_full_name()]["value" + FIELD_MARKER] == "updated" + + @pytest.mark.asyncio async def test_linked_state_patch_preserves_interval_computed_refresh(): """Interval computed vars refreshed during patching must be emitted."""