From 63c7d0f9836a6c720c1d2ecfa6ce76bcd21d5e95 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 8 Oct 2026 12:04:22 +0530 Subject: [PATCH 01/14] Avoid duplicate storage recomputation during hydration Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- news/+7508.bugfix.md | 1 + reflex/state.py | 25 ++++++++++++++++++++++--- 2 files changed, 23 insertions(+), 3 deletions(-) create mode 100644 news/+7508.bugfix.md diff --git a/news/+7508.bugfix.md b/news/+7508.bugfix.md new file mode 100644 index 00000000000..1a95f91d041 --- /dev/null +++ b/news/+7508.bugfix.md @@ -0,0 +1 @@ +Avoid redundant recomputation of storage-dependent computed vars during page hydration. diff --git a/reflex/state.py b/reflex/state.py index 369b381db77..ece58036ebc 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -2415,11 +2415,30 @@ async def hydrate_and_load( await ctx.emit_delta(delta=delta) # Follow-up corrections must be allowed to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) - # The browser's values only: the reset defaults stay clean, so they are - # not written back to the browser. + # Let get_delta overrides inspect browser values without invalidating + # their computed dependents. Re-resolve only when an override changes + # a value, so dependent computed vars reflect that correction. for var_state, var_name in applied: var_state.dirty_vars.add(var_name) - var_state._mark_dirty((var_name,)) + var_state._mark_ancestors_dirty() + if applied: + correction_delta = await self._get_resolved_delta() + dependent_deltas = [] + for var_state, var_name in applied: + key = var_name + FIELD_MARKER + subdelta = correction_delta.get(var_state.get_full_name(), {}) + if key in subdelta and subdelta[key] != var_state.get_value(var_name): + setattr(var_state, var_name, subdelta[key]) + var_state._mark_dirty((var_name,)) + dependent_deltas.append( + await _resolve_delta(BaseState.get_delta(var_state)) + ) + for dependent_delta in dependent_deltas: + for state_name, changes in dependent_delta.items(): + correction_delta.setdefault(state_name, {}).update(changes) + if ctx.emit_delta_impl is not None and correction_delta: + await ctx.emit_delta(delta=correction_delta) + self._clean() if not RegistrationContext.get().app.get_load_events(self.rx_router_url.path): self.is_hydrated = True return None From 520458c9a728c6b2ebad9a28314cde97adaafe33 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 8 Oct 2026 14:35:38 +0530 Subject: [PATCH 02/14] Fix hydration computed var reconciliation --- reflex/istate/delta.py | 26 ++++- reflex/state.py | 42 +++++--- .../processor/test_base_state_processor.py | 102 +++++++++++++++++- 3 files changed, 152 insertions(+), 18 deletions(-) diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index 9ec87aac3d4..265bd0ca911 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -40,6 +40,13 @@ "_record_delta_values", default=True ) +# During the client-storage reconciliation probe, dirty storage vars must reach +# get_delta overrides without invalidating their computed dependents. Expired +# and always-dirty computed vars are still processed by build_delta. +_suppress_delta_dependency_invalidation: ContextVar[bool] = ContextVar( + "_suppress_delta_dependency_invalidation", default=False +) + class _DeltaRecord(NamedTuple): """An uncached var value that counts as sent once the delta delivers it.""" @@ -82,6 +89,20 @@ def _suppress_delta_recording() -> Iterator[None]: _record_delta_values.reset(token) +@contextlib.contextmanager +def _suppress_computed_var_dependency_invalidation() -> Iterator[None]: + """Suppress computed-var invalidation caused by dirty source vars. + + Yields: + None, with computed-var dependency invalidation suppressed. + """ + token = _suppress_delta_dependency_invalidation.set(True) + try: + yield + finally: + _suppress_delta_dependency_invalidation.reset(token) + + def _commit_delta_records(pending: list[_DeltaRecord], delta: Delta) -> None: """Record what a delivered delta leaves the client holding for each value. @@ -222,7 +243,10 @@ def build_delta(state: BaseState) -> Delta: """ delta = {} - state._mark_dirty_computed_vars() + if _suppress_delta_dependency_invalidation.get(): + state._mark_dirty_computed_vars(()) + else: + state._mark_dirty_computed_vars() delta_vars = state.dirty_vars & state._frontend_var_names always_dirty_computed_vars = state._always_dirty_computed_vars diff --git a/reflex/state.py b/reflex/state.py index ece58036ebc..f5b23ef3b4a 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -88,6 +88,8 @@ Delta, DeltaMapping, _resolve_delta, + _suppress_computed_var_dependency_invalidation, + _suppress_delta_dependency_invalidation, build_delta, clean_state, resolve_delta, @@ -1943,7 +1945,10 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: var_names: The vars of this state that changed; all its dirty vars if omitted. """ self._mark_ancestors_dirty() - self._mark_dirty_computed_vars(var_names) + if _suppress_delta_dependency_invalidation.get(): + self._mark_dirty_computed_vars(()) + else: + self._mark_dirty_computed_vars(var_names) def _mark_ancestors_dirty(self) -> None: """Record this state as a dirty substate of each of its ancestors.""" @@ -2398,7 +2403,8 @@ async def hydrate_and_load( from reflex_base.event.context import EventContext self._reset_client_storage() - applied = await _apply_client_storage_vars(self, vars) if vars else [] + with _suppress_computed_var_dependency_invalidation(): + applied = await _apply_client_storage_vars(self, vars) if vars else [] self._clean() # The snapshot must carry is_hydrated=False: the frontend skips # writing client storage for a delta that is not yet hydrated, and @@ -2415,26 +2421,36 @@ async def hydrate_and_load( await ctx.emit_delta(delta=delta) # Follow-up corrections must be allowed to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) - # Let get_delta overrides inspect browser values without invalidating - # their computed dependents. Re-resolve only when an override changes - # a value, so dependent computed vars reflect that correction. + # Probe overrides with browser values, but don't invalidate computed + # dependents unless an override actually changes a storage value. for var_state, var_name in applied: var_state.dirty_vars.add(var_name) var_state._mark_ancestors_dirty() if applied: - correction_delta = await self._get_resolved_delta() - dependent_deltas = [] + with _suppress_computed_var_dependency_invalidation(): + correction_delta = await self._get_resolved_delta() + changed_vars = [] for var_state, var_name in applied: key = var_name + FIELD_MARKER subdelta = correction_delta.get(var_state.get_full_name(), {}) - if key in subdelta and subdelta[key] != var_state.get_value(var_name): + browser_value = var_state.get_value(var_name) + overridden = key in subdelta and _serialize_var( + subdelta[key] + ) != _serialize_var(browser_value) + if overridden: setattr(var_state, var_name, subdelta[key]) + changed_vars.append((var_state, var_name)) + if changed_vars: + # Discard the probe's dirtiness before producing a full-tree + # delta. This makes dependencies on sibling states visible and + # prevents expired/always-dirty computed vars from being + # evaluated a second time in the correction pass. + self._clean() + for var_state, var_name in changed_vars: + var_state.dirty_vars.add(var_name) var_state._mark_dirty((var_name,)) - dependent_deltas.append( - await _resolve_delta(BaseState.get_delta(var_state)) - ) - for dependent_delta in dependent_deltas: - for state_name, changes in dependent_delta.items(): + corrected_delta = await self._get_resolved_delta() + for state_name, changes in corrected_delta.items(): correction_delta.setdefault(state_name, {}).update(changes) if ctx.emit_delta_impl is not None and correction_delta: await ctx.emit_delta(delta=correction_delta) diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index b26200917a1..44d4cc6666e 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -1907,6 +1907,7 @@ def load(self): @pytest.mark.parametrize("with_load", [True, False]) @pytest.mark.parametrize("with_hashes", [True, False]) +@pytest.mark.parametrize("client_value", ["stale", "fresh"]) @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) @@ -1917,8 +1918,9 @@ async def test_hydrate_passes_client_storage_through_get_delta( token: str, with_load: bool, with_hashes: bool, + client_value: str, ): - """A get_delta override sees the browser's client storage at boot, once. + """A get_delta override reconciles browser storage without redundant recomputation. Overrides reconcile client storage there (reflex-enterprise checks its OIDC token hash), so the boot values must reach them as @@ -1932,9 +1934,11 @@ async def test_hydrate_passes_client_storage_through_get_delta( token: The client token. with_load: Whether the page has an on-load handler. with_hashes: Whether this is the first boot or a reconnect. + client_value: The browser-provided value to reconcile. """ key = "token_hash" + FIELD_MARKER seen: list[str] = [] + computed_values: list[str] = [] class ReconcileState(State): token_hash: str = rx.LocalStorage("") @@ -1959,16 +1963,35 @@ def get_delta(self) -> Delta: def load(self): """Provide an on-load event.""" + class DependentState(State): + """A sibling state with a cached var depending on client storage.""" + + @rx.var(cache=True, auto_deps=False) + async def token_hash_copy(self) -> str: + storage_state = await self.get_state(ReconcileState) + computed_values.append(storage_state.token_hash) + return storage_state.token_hash + + DependentState.computed_vars["token_hash_copy"].add_dependency( + DependentState, + ReconcileState.token_hash, # pyright: ignore[reportArgumentType] + ) + wired_app.add_page( - lambda: rx.text(ReconcileState.token_hash, ReconcileState.unset), + lambda: rx.text( + ReconcileState.token_hash, + ReconcileState.unset, + DependentState.token_hash_copy, + ), route="/", on_load=ReconcileState.load if with_load else None, ) wired_app._compile_page("index") name = ReconcileState.get_full_name() - payload: dict[str, Any] = {"vars": {f"{name}.{key}": "stale"}} + payload: dict[str, Any] = {"vars": {f"{name}.{key}": client_value}} if with_hashes: payload["hashes"] = state_snapshot_hashes(compile_state(State)) + computed_values.clear() boot_name = Event.from_event_type(State.hydrate_and_load())[0].name # pyright: ignore[reportCallIssue] async with real_base_state_processor as processor: @@ -1976,7 +1999,7 @@ def load(self): await processor.enqueue(token, _boot_event(boot_name, payload)) ).wait_all() - assert seen == ["stale"] + assert seen == (["stale", "fresh"] if client_value == "stale" else ["fresh"]) hydrated_key = CompileVars.IS_HYDRATED + FIELD_MARKER assert emitted_deltas[0][1][State.get_full_name()][hydrated_key] is False persisted = [ @@ -1987,6 +2010,77 @@ def load(self): ] assert any(subdelta.get(key) == "fresh" for subdelta in persisted) assert not any("unset" + FIELD_MARKER in subdelta for subdelta in persisted) + dependent_key = "token_hash_copy" + FIELD_MARKER + assert any( + delta.get(DependentState.get_full_name(), {}).get(dependent_key) == "fresh" + for _, delta in emitted_deltas + ) + assert computed_values == ( + ["stale", "fresh"] if client_value == "stale" else ["fresh"] + ) + + +@pytest.mark.parametrize( + "processor_state_manager", ["in_process", "redis"], indirect=True +) +async def test_hydrate_reconciles_json_distinct_storage_values( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], + token: str, +): + """Hydration treats JSON-distinct values as changes even when Python equates them. + + Args: + wired_app: The app wired to the state manager. + real_base_state_processor: The event processor. + emitted_deltas: Captured deltas. + token: The client token. + """ + + class StorageState(State): + value: int | bool = rx.LocalStorage(1) # pyright: ignore[reportAssignmentType] + + @_override_base_method + def get_delta(self) -> Delta: + delta = super().get_delta() + subdelta = delta.get(self.get_full_name(), {}) + key = "value" + FIELD_MARKER + if subdelta.get(key) == 1: + subdelta[key] = True + return delta + + class DependentState(State): + @rx.var(cache=True, auto_deps=False) + async def value_type(self) -> str: + storage_state = await self.get_state(StorageState) + return type(storage_state.value).__name__ + + DependentState.computed_vars["value_type"].add_dependency( + DependentState, + StorageState.value, # pyright: ignore[reportArgumentType] + ) + wired_app.add_page( + lambda: rx.text(StorageState.value, DependentState.value_type), route="/" + ) + wired_app._compile_page("index") + name = StorageState.get_full_name() + payload: dict[str, Any] = {"vars": {f"{name}.value{FIELD_MARKER}": 1}} + boot_name = Event.from_event_type(State.hydrate_and_load())[0].name # pyright: ignore[reportCallIssue] + + async with real_base_state_processor as processor: + await ( + await processor.enqueue(token, _boot_event(boot_name, payload)) + ).wait_all() + + value_type_key = "value_type" + FIELD_MARKER + assert any( + delta.get(DependentState.get_full_name(), {}).get(value_type_key) == "bool" + for _, delta in emitted_deltas + ) + async with _read_back(real_base_state_processor, token) as root: + storage_state = await root.get_state(StorageState) + assert storage_state.value is True @pytest.mark.parametrize( From 7158ab0ca919f13388dd1cceb885bfd2869b9478 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 8 Oct 2026 15:06:45 +0530 Subject: [PATCH 03/14] Limit hydration dependency suppression to storage vars --- reflex/istate/delta.py | 50 +++++++++++++++---- reflex/state.py | 28 +++++++---- .../processor/test_base_state_processor.py | 19 +++++++ 3 files changed, 76 insertions(+), 21 deletions(-) diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index 265bd0ca911..064ab8b7f1f 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -6,7 +6,7 @@ import contextlib import inspect import time -from collections.abc import Coroutine, Iterator, Mapping +from collections.abc import Coroutine, Iterable, Iterator, Mapping from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Final, NamedTuple @@ -43,8 +43,8 @@ # During the client-storage reconciliation probe, dirty storage vars must reach # get_delta overrides without invalidating their computed dependents. Expired # and always-dirty computed vars are still processed by build_delta. -_suppress_delta_dependency_invalidation: ContextVar[bool] = ContextVar( - "_suppress_delta_dependency_invalidation", default=False +_suppressed_delta_dependency_sources: ContextVar[frozenset[tuple[int, str]]] = ( + ContextVar("_suppressed_delta_dependency_sources", default=frozenset()) ) @@ -90,17 +90,46 @@ def _suppress_delta_recording() -> Iterator[None]: @contextlib.contextmanager -def _suppress_computed_var_dependency_invalidation() -> Iterator[None]: - """Suppress computed-var invalidation caused by dirty source vars. +def _suppress_computed_var_dependency_invalidation( + sources: frozenset[tuple[int, str]], +) -> Iterator[None]: + """Suppress computed-var invalidation for selected state vars. + + Args: + sources: State instance ids and var names whose dependents should not be + invalidated during the traversal. Yields: None, with computed-var dependency invalidation suppressed. """ - token = _suppress_delta_dependency_invalidation.set(True) + token = _suppressed_delta_dependency_sources.set(sources) try: yield finally: - _suppress_delta_dependency_invalidation.reset(token) + _suppressed_delta_dependency_sources.reset(token) + + +def _get_unsuppressed_var_names( + state: BaseState, var_names: Iterable[str] | None +) -> Iterable[str] | None: + """Exclude selected source vars from computed-var invalidation. + + Args: + state: The state whose dependencies are being invalidated. + var_names: The changed var names, or None to use all dirty vars. + + Returns: + The var names that should invalidate computed dependencies. + """ + suppressed = _suppressed_delta_dependency_sources.get() + if not suppressed: + return var_names + excluded = {name for state_id, name in suppressed if state_id == id(state)} + return ( + state.dirty_vars - excluded + if var_names is None + else (name for name in var_names if name not in excluded) + ) def _commit_delta_records(pending: list[_DeltaRecord], delta: Delta) -> None: @@ -243,10 +272,9 @@ def build_delta(state: BaseState) -> Delta: """ delta = {} - if _suppress_delta_dependency_invalidation.get(): - state._mark_dirty_computed_vars(()) - else: - state._mark_dirty_computed_vars() + state._mark_dirty_computed_vars( + _get_unsuppressed_var_names(state, state.dirty_vars) + ) delta_vars = state.dirty_vars & state._frontend_var_names always_dirty_computed_vars = state._always_dirty_computed_vars diff --git a/reflex/state.py b/reflex/state.py index f5b23ef3b4a..7c7865b8a65 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -89,7 +89,6 @@ DeltaMapping, _resolve_delta, _suppress_computed_var_dependency_invalidation, - _suppress_delta_dependency_invalidation, build_delta, clean_state, resolve_delta, @@ -1945,10 +1944,7 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: var_names: The vars of this state that changed; all its dirty vars if omitted. """ self._mark_ancestors_dirty() - if _suppress_delta_dependency_invalidation.get(): - self._mark_dirty_computed_vars(()) - else: - self._mark_dirty_computed_vars(var_names) + self._mark_dirty_computed_vars(var_names) def _mark_ancestors_dirty(self) -> None: """Record this state as a dirty substate of each of its ancestors.""" @@ -2403,8 +2399,7 @@ async def hydrate_and_load( from reflex_base.event.context import EventContext self._reset_client_storage() - with _suppress_computed_var_dependency_invalidation(): - applied = await _apply_client_storage_vars(self, vars) if vars else [] + applied = await _apply_client_storage_vars(self, vars) if vars else [] self._clean() # The snapshot must carry is_hydrated=False: the frontend skips # writing client storage for a delta that is not yet hydrated, and @@ -2427,8 +2422,21 @@ async def hydrate_and_load( var_state.dirty_vars.add(var_name) var_state._mark_ancestors_dirty() if applied: - with _suppress_computed_var_dependency_invalidation(): + dirty_before_probe: dict[int, tuple[BaseState, set[str]]] = {} + states_to_check: list[BaseState] = [self] + while states_to_check: + state = states_to_check.pop() + dirty_before_probe[id(state)] = (state, set(state.dirty_vars)) + states_to_check.extend(state.substates.values()) + suppressed_sources = frozenset((id(state), name) for state, name in applied) + with _suppress_computed_var_dependency_invalidation(suppressed_sources): correction_delta = await self._get_resolved_delta() + mutated_vars = [ + (state, name) + for state, dirty_vars in dirty_before_probe.values() + for name in state.dirty_vars - dirty_vars + if name not in state.computed_vars + ] changed_vars = [] for var_state, var_name in applied: key = var_name + FIELD_MARKER @@ -2440,13 +2448,13 @@ async def hydrate_and_load( if overridden: setattr(var_state, var_name, subdelta[key]) changed_vars.append((var_state, var_name)) - if changed_vars: + if changed_vars or mutated_vars: # Discard the probe's dirtiness before producing a full-tree # delta. This makes dependencies on sibling states visible and # prevents expired/always-dirty computed vars from being # evaluated a second time in the correction pass. self._clean() - for var_state, var_name in changed_vars: + for var_state, var_name in [*changed_vars, *mutated_vars]: var_state.dirty_vars.add(var_name) var_state._mark_dirty((var_name,)) corrected_delta = await self._get_resolved_delta() diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index 44d4cc6666e..207af457b13 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -1943,6 +1943,7 @@ async def test_hydrate_passes_client_storage_through_get_delta( class ReconcileState(State): token_hash: str = rx.LocalStorage("") unset: str = rx.Cookie("unset-default") + status: str = "initial" @_override_base_method def get_delta(self) -> Delta: @@ -1957,6 +1958,7 @@ def get_delta(self) -> Delta: seen.append(value) if value == "stale": subdelta[key] = "fresh" + self.status = "reconciled" return delta @event @@ -1972,16 +1974,26 @@ async def token_hash_copy(self) -> str: computed_values.append(storage_state.token_hash) return storage_state.token_hash + @rx.var(cache=True, auto_deps=False) + async def status_copy(self) -> str: + storage_state = await self.get_state(ReconcileState) + return storage_state.status + DependentState.computed_vars["token_hash_copy"].add_dependency( DependentState, ReconcileState.token_hash, # pyright: ignore[reportArgumentType] ) + DependentState.computed_vars["status_copy"].add_dependency( + DependentState, + ReconcileState.status, # pyright: ignore[reportArgumentType] + ) wired_app.add_page( lambda: rx.text( ReconcileState.token_hash, ReconcileState.unset, DependentState.token_hash_copy, + DependentState.status_copy, ), route="/", on_load=ReconcileState.load if with_load else None, @@ -2018,6 +2030,13 @@ async def token_hash_copy(self) -> str: assert computed_values == ( ["stale", "fresh"] if client_value == "stale" else ["fresh"] ) + status_key = "status_copy" + FIELD_MARKER + if client_value == "stale": + assert any( + delta.get(DependentState.get_full_name(), {}).get(status_key) + == "reconciled" + for _, delta in emitted_deltas + ) @pytest.mark.parametrize( From a7bf25f4f3e07afa9f69e42a0212b787a25a2759 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 12:02:47 +0530 Subject: [PATCH 04/14] Reduce hydration delta reconciliation overhead --- reflex/istate/delta.py | 18 ++++++---- reflex/state.py | 77 ++++++++++++++++++++++++++++++------------ 2 files changed, 66 insertions(+), 29 deletions(-) diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index 064ab8b7f1f..960582fd15e 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -110,20 +110,20 @@ def _suppress_computed_var_dependency_invalidation( def _get_unsuppressed_var_names( - state: BaseState, var_names: Iterable[str] | None + state: BaseState, + var_names: Iterable[str] | None, + suppressed: frozenset[tuple[int, str]], ) -> Iterable[str] | None: """Exclude selected source vars from computed-var invalidation. Args: state: The state whose dependencies are being invalidated. var_names: The changed var names, or None to use all dirty vars. + suppressed: The state vars excluded from dependency invalidation. Returns: The var names that should invalidate computed dependencies. """ - suppressed = _suppressed_delta_dependency_sources.get() - if not suppressed: - return var_names excluded = {name for state_id, name in suppressed if state_id == id(state)} return ( state.dirty_vars - excluded @@ -272,9 +272,13 @@ def build_delta(state: BaseState) -> Delta: """ delta = {} - state._mark_dirty_computed_vars( - _get_unsuppressed_var_names(state, state.dirty_vars) - ) + suppressed_sources = _suppressed_delta_dependency_sources.get() + if suppressed_sources: + state._mark_dirty_computed_vars( + _get_unsuppressed_var_names(state, state.dirty_vars, suppressed_sources) + ) + else: + state._mark_dirty_computed_vars() delta_vars = state.dirty_vars & state._frontend_var_names always_dirty_computed_vars = state._always_dirty_computed_vars diff --git a/reflex/state.py b/reflex/state.py index 7c7865b8a65..33d9867b89d 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -13,6 +13,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 @@ -99,6 +100,31 @@ from reflex.utils import console, format, types from reflex.utils.exec import is_testing_env +_dirty_var_collector: ContextVar[set[tuple[BaseState, str]] | None] = ContextVar( + "_dirty_var_collector", default=None +) + + +@contextlib.contextmanager +def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: + """Collect vars dirtied by assignments in the current context. + + Yields: + The set of state vars dirtied while the context is active. + """ + dirty_vars: set[tuple[BaseState, str]] = set() + token = _dirty_var_collector.set(dirty_vars) + try: + yield dirty_vars + finally: + _dirty_var_collector.reset(token) + + +def _state_ancestors(state: BaseState) -> Iterator[BaseState]: + """Yield the parents of a state, from nearest to root.""" + while (state := state.parent_state) is not None: + yield state + @functools.cache def _stale_pickle_keys(cls: type[BaseState]) -> frozenset[str]: @@ -1943,6 +1969,12 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: Args: var_names: The vars of this state that changed; all its dirty vars if omitted. """ + collector = _dirty_var_collector.get() + if collector is not None: + var_names = tuple(self.dirty_vars if var_names is None else var_names) + collector.update( + (self, name) for name in var_names if name not in self.computed_vars + ) self._mark_ancestors_dirty() self._mark_dirty_computed_vars(var_names) @@ -2416,27 +2448,29 @@ async def hydrate_and_load( await ctx.emit_delta(delta=delta) # Follow-up corrections must be allowed to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) - # Probe overrides with browser values, but don't invalidate computed - # dependents unless an override actually changes a storage value. - for var_state, var_name in applied: - var_state.dirty_vars.add(var_name) - var_state._mark_ancestors_dirty() if applied: - dirty_before_probe: dict[int, tuple[BaseState, set[str]]] = {} - states_to_check: list[BaseState] = [self] - while states_to_check: - state = states_to_check.pop() - dirty_before_probe[id(state)] = (state, set(state.dirty_vars)) - states_to_check.extend(state.substates.values()) + storage_states = {id(state): state for state, _ in applied} + probe_root = any( + type(ancestor).get_delta is not BaseState.get_delta + for state in storage_states.values() + for ancestor in _state_ancestors(state) + ) + # Parent overrides wrap child deltas, so keep their root traversal. + if probe_root: + storage_states = {id(self): self} + for var_state, var_name in applied: + var_state.dirty_vars.add(var_name) + var_state._mark_ancestors_dirty() suppressed_sources = frozenset((id(state), name) for state, name in applied) - with _suppress_computed_var_dependency_invalidation(suppressed_sources): - correction_delta = await self._get_resolved_delta() - mutated_vars = [ - (state, name) - for state, dirty_vars in dirty_before_probe.values() - for name in state.dirty_vars - dirty_vars - if name not in state.computed_vars - ] + with ( + _collect_dirty_vars() as mutated_vars, + _suppress_computed_var_dependency_invalidation(suppressed_sources), + ): + correction_delta: Delta = {} + for state in storage_states.values(): + state_delta = await state._get_resolved_delta() + for state_name, changes in state_delta.items(): + correction_delta.setdefault(state_name, {}).update(changes) changed_vars = [] for var_state, var_name in applied: key = var_name + FIELD_MARKER @@ -2450,9 +2484,8 @@ async def hydrate_and_load( changed_vars.append((var_state, var_name)) if changed_vars or mutated_vars: # Discard the probe's dirtiness before producing a full-tree - # delta. This makes dependencies on sibling states visible and - # prevents expired/always-dirty computed vars from being - # evaluated a second time in the correction pass. + # delta. This makes dependencies on sibling states visible + # and avoids evaluating interval vars in the probe again. self._clean() for var_state, var_name in [*changed_vars, *mutated_vars]: var_state.dirty_vars.add(var_name) From 8046dc6071b04743f5c05beeecdfbf77bba48a60 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 12:54:34 +0530 Subject: [PATCH 05/14] Avoid repeated child hydration probes --- reflex/state.py | 10 +++- .../processor/test_base_state_processor.py | 52 +++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/reflex/state.py b/reflex/state.py index 33d9867b89d..f106038118b 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -2449,7 +2449,15 @@ async def hydrate_and_load( # Follow-up corrections must be allowed to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) if applied: - storage_states = {id(state): state for state, _ in applied} + applied_states = {id(state): state for state, _ in applied} + storage_states = { + state_id: state + for state_id, state in applied_states.items() + if not any( + id(ancestor) in applied_states + for ancestor in _state_ancestors(state) + ) + } probe_root = any( type(ancestor).get_delta is not BaseState.get_delta for state in storage_states.values() diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index 207af457b13..f9cf0913047 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -2102,6 +2102,58 @@ async def value_type(self) -> str: assert storage_state.value is True +async def test_hydrate_probes_nested_storage_states_once( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + token: str, +): + """A parent probe already visits child states that also own storage vars. + + Args: + wired_app: The app wired to the state manager. + real_base_state_processor: The event processor. + token: The client token. + """ + child_storage_probes = 0 + + class ParentStorageState(State): + parent_value: str = rx.LocalStorage("") + + class ChildStorageState(ParentStorageState): + child_value: str = rx.LocalStorage("") + + @rx.var(cache=False) + def uncached_value(self) -> str: + return self.child_value + + @_override_base_method + def get_delta(self) -> Delta: + nonlocal child_storage_probes + if "child_value" in self.dirty_vars: + child_storage_probes += 1 + return super().get_delta() + + wired_app.add_page( + lambda: rx.text(ParentStorageState.parent_value, ChildStorageState.child_value), + route="/", + ) + wired_app._compile_page("index") + boot_name = Event.from_event_type(State.hydrate_and_load())[0].name # pyright: ignore[reportCallIssue] + payload = { + "vars": { + f"{ParentStorageState.get_full_name()}.parent_value{FIELD_MARKER}": "parent", + f"{ChildStorageState.get_full_name()}.child_value{FIELD_MARKER}": "child", + } + } + + async with real_base_state_processor as processor: + await ( + await processor.enqueue(token, _boot_event(boot_name, payload)) + ).wait_all() + + assert child_storage_probes == 1 + + @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) From e3d7ead246b97addc0699a42572bfd23cc122777 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 13:42:36 +0530 Subject: [PATCH 06/14] Scope hydration correction to dirty branches --- reflex/state.py | 22 ++++++- .../processor/test_base_state_processor.py | 65 +++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/reflex/state.py b/reflex/state.py index f106038118b..fb75ce3c532 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -126,6 +126,15 @@ def _state_ancestors(state: BaseState) -> Iterator[BaseState]: yield state +def _dirty_delta_roots(state: BaseState) -> Iterator[BaseState]: + """Yield the highest dirty states whose deltas include their dirty descendants.""" + if state.dirty_vars: + yield state + return + for substate_name in state.dirty_substates: + yield from _dirty_delta_roots(state.substates[substate_name]) + + @functools.cache def _stale_pickle_keys(cls: type[BaseState]) -> frozenset[str]: """Get the keys of pickles that are not restored into the instance dict. @@ -2498,7 +2507,18 @@ async def hydrate_and_load( for var_state, var_name in [*changed_vars, *mutated_vars]: var_state.dirty_vars.add(var_name) var_state._mark_dirty((var_name,)) - corrected_delta = await self._get_resolved_delta() + correction_states = list(_dirty_delta_roots(self)) + if any( + type(ancestor).get_delta is not BaseState.get_delta + for state in correction_states + for ancestor in _state_ancestors(state) + ): + correction_states = [self] + corrected_delta: Delta = {} + for state in correction_states: + state_delta = await state._get_resolved_delta() + for state_name, changes in state_delta.items(): + corrected_delta.setdefault(state_name, {}).update(changes) for state_name, changes in corrected_delta.items(): correction_delta.setdefault(state_name, {}).update(changes) if ctx.emit_delta_impl is not None and correction_delta: diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index f9cf0913047..dc2b5c5b28d 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -2154,6 +2154,71 @@ def get_delta(self) -> Delta: assert child_storage_probes == 1 +async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + token: str, +): + """A stale storage correction does not reevaluate unrelated state branches. + + Args: + wired_app: The app wired to the state manager. + real_base_state_processor: The event processor. + token: The client token. + """ + unrelated_computations = 0 + + class StorageState(State): + value: str = rx.LocalStorage("") + + @_override_base_method + def get_delta(self) -> Delta: + delta = super().get_delta() + key = "value" + FIELD_MARKER + subdelta = delta.get(self.get_full_name(), {}) + if subdelta.get(key) == "stale": + subdelta[key] = "fresh" + return delta + + class UnrelatedState(State): + @rx.var(cache=False) + def value(self) -> str: + nonlocal unrelated_computations + unrelated_computations += 1 + return "unchanged" + + wired_app.add_page( + lambda: rx.text(StorageState.value, UnrelatedState.value), route="/" + ) + wired_app._compile_page("index") + boot_name = Event.from_event_type(State.hydrate_and_load())[0].name # pyright: ignore[reportCallIssue] + + async def hydrate(value: str) -> int: + """Hydrate with a storage value and return unrelated computations. + + Args: + value: The browser-provided storage value. + + Returns: + The number of unrelated computed-var evaluations. + """ + nonlocal unrelated_computations + unrelated_computations = 0 + payload = { + "vars": {f"{StorageState.get_full_name()}.value{FIELD_MARKER}": value} + } + async with real_base_state_processor as processor: + await ( + await processor.enqueue(token, _boot_event(boot_name, payload)) + ).wait_all() + return unrelated_computations + + fresh_computations = await hydrate("fresh") + stale_computations = await hydrate("stale") + + assert stale_computations == fresh_computations + + @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) From 3fcf24a1a237197afafd5c6b20cc73360336966f Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 13:46:01 +0530 Subject: [PATCH 07/14] Fix state ancestor typing --- reflex/state.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/reflex/state.py b/reflex/state.py index fb75ce3c532..e25682c735c 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -122,8 +122,10 @@ def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: def _state_ancestors(state: BaseState) -> Iterator[BaseState]: """Yield the parents of a state, from nearest to root.""" - while (state := state.parent_state) is not None: - yield state + parent = state.parent_state + while parent is not None: + yield parent + parent = parent.parent_state def _dirty_delta_roots(state: BaseState) -> Iterator[BaseState]: From 0b1bd7db6cbdc6b8eeadad2bb0dd2e9833a86804 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 14:12:14 +0530 Subject: [PATCH 08/14] Avoid collector lookup on ordinary state writes --- reflex/state.py | 17 +++++++++++------ .../processor/test_base_state_processor.py | 8 ++++++++ 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/reflex/state.py b/reflex/state.py index e25682c735c..09ff099821a 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -103,6 +103,7 @@ _dirty_var_collector: ContextVar[set[tuple[BaseState, str]] | None] = ContextVar( "_dirty_var_collector", default=None ) +_dirty_var_collection_depth = 0 @contextlib.contextmanager @@ -112,12 +113,15 @@ def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: Yields: The set of state vars dirtied while the context is active. """ + global _dirty_var_collection_depth dirty_vars: set[tuple[BaseState, str]] = set() + _dirty_var_collection_depth += 1 token = _dirty_var_collector.set(dirty_vars) try: yield dirty_vars finally: _dirty_var_collector.reset(token) + _dirty_var_collection_depth -= 1 def _state_ancestors(state: BaseState) -> Iterator[BaseState]: @@ -1980,12 +1984,13 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: Args: var_names: The vars of this state that changed; all its dirty vars if omitted. """ - collector = _dirty_var_collector.get() - if collector is not None: - var_names = tuple(self.dirty_vars if var_names is None else var_names) - collector.update( - (self, name) for name in var_names if name not in self.computed_vars - ) + if _dirty_var_collection_depth: + collector = _dirty_var_collector.get() + if collector is not None: + var_names = tuple(self.dirty_vars if var_names is None else var_names) + collector.update( + (self, name) for name in var_names if name not in self.computed_vars + ) self._mark_ancestors_dirty() self._mark_dirty_computed_vars(var_names) diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index dc2b5c5b28d..8db10890429 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -2157,6 +2157,7 @@ def get_delta(self) -> Delta: async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( wired_app: App, real_base_state_processor: BaseStateEventProcessor, + emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): """A stale storage correction does not reevaluate unrelated state branches. @@ -2164,6 +2165,7 @@ async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( Args: wired_app: The app wired to the state manager. real_base_state_processor: The event processor. + emitted_deltas: Captured deltas. token: The client token. """ unrelated_computations = 0 @@ -2214,9 +2216,15 @@ async def hydrate(value: str) -> int: return unrelated_computations fresh_computations = await hydrate("fresh") + emitted_deltas.clear() stale_computations = await hydrate("stale") assert stale_computations == fresh_computations + assert any( + delta.get(StorageState.get_full_name(), {}).get("value" + FIELD_MARKER) + == "fresh" + for _, delta in emitted_deltas + ) @pytest.mark.parametrize( From b4746bc895c5481cd81f3becc875197fc203fb55 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Fri, 9 Oct 2026 14:22:34 +0530 Subject: [PATCH 09/14] Synchronize hydration collector counter --- reflex/state.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/reflex/state.py b/reflex/state.py index 09ff099821a..ed8b8fbf128 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -12,6 +12,7 @@ import pickle import re import sys +import threading from collections.abc import Callable, Iterable, Iterator, Sequence from contextvars import ContextVar from datetime import timedelta @@ -104,6 +105,7 @@ "_dirty_var_collector", default=None ) _dirty_var_collection_depth = 0 +_dirty_var_collection_lock = threading.Lock() @contextlib.contextmanager @@ -115,13 +117,15 @@ def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: """ global _dirty_var_collection_depth dirty_vars: set[tuple[BaseState, str]] = set() - _dirty_var_collection_depth += 1 + with _dirty_var_collection_lock: + _dirty_var_collection_depth += 1 token = _dirty_var_collector.set(dirty_vars) try: yield dirty_vars finally: _dirty_var_collector.reset(token) - _dirty_var_collection_depth -= 1 + with _dirty_var_collection_lock: + _dirty_var_collection_depth -= 1 def _state_ancestors(state: BaseState) -> Iterator[BaseState]: From 62f0c04497a219fafd22ad53a3f36c7b635e62eb Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 13 Oct 2026 11:24:04 +0530 Subject: [PATCH 10/14] chore: apply generated cleanup for PR #7515 --- pyi_hashes.json | 2 +- reflex/istate/delta.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyi_hashes.json b/pyi_hashes.json index e96179792f8..61656d31c61 100644 --- a/pyi_hashes.json +++ b/pyi_hashes.json @@ -80,7 +80,7 @@ "packages/reflex-components-radix/src/reflex_components_radix/themes/components/radio_group.pyi": "e933b49dab8027e1829cdd90ba99cc5d", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/scroll_area.pyi": "b83dd81a37aaad75dcb6341e94e6d41f", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/segmented_control.pyi": "dfdde032961bda7905ef4256eb19152d", - "packages/reflex-components-radix/src/reflex_components_radix/themes/components/select.pyi": "98818bc17dd54d4a0bfb48f7d971875a", + "packages/reflex-components-radix/src/reflex_components_radix/themes/components/select.pyi": "75f5da11fe90491751d8ebcb3d1332a1", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/separator.pyi": "b97507dd3c8ee6464b733fb61558c0ba", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/skeleton.pyi": "77e709d2f460b19617c9fc2e45537e18", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/slider.pyi": "27ae71e52b3a6abce5d94e83c354afb9", diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index 86473327e53..f140402c2dc 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -6,7 +6,7 @@ import contextlib import inspect import time -from collections.abc import Coroutine, Iterable, Iterator, Mapping +from collections.abc import Iterable, Iterator, Mapping from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Final, NamedTuple From fa45fd7d2167e25fb8db5da35f3d538ae2fbaa88 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 13 Oct 2026 11:34:41 +0530 Subject: [PATCH 11/14] fix: keep generated stub hash consistent with CI --- pyi_hashes.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyi_hashes.json b/pyi_hashes.json index 61656d31c61..e96179792f8 100644 --- a/pyi_hashes.json +++ b/pyi_hashes.json @@ -80,7 +80,7 @@ "packages/reflex-components-radix/src/reflex_components_radix/themes/components/radio_group.pyi": "e933b49dab8027e1829cdd90ba99cc5d", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/scroll_area.pyi": "b83dd81a37aaad75dcb6341e94e6d41f", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/segmented_control.pyi": "dfdde032961bda7905ef4256eb19152d", - "packages/reflex-components-radix/src/reflex_components_radix/themes/components/select.pyi": "75f5da11fe90491751d8ebcb3d1332a1", + "packages/reflex-components-radix/src/reflex_components_radix/themes/components/select.pyi": "98818bc17dd54d4a0bfb48f7d971875a", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/separator.pyi": "b97507dd3c8ee6464b733fb61558c0ba", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/skeleton.pyi": "77e709d2f460b19617c9fc2e45537e18", "packages/reflex-components-radix/src/reflex_components_radix/themes/components/slider.pyi": "27ae71e52b3a6abce5d94e83c354afb9", From ef565faf1fab6b7f7485a4f035909956bc8fbe7a Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 11:15:20 +0530 Subject: [PATCH 12/14] Preserve storage writes from hydration overrides --- .github/workflows/examples.yml | 2 +- reflex/state.py | 6 ++ .../processor/test_base_state_processor.py | 56 +++++++++++++++++++ 3 files changed, 63 insertions(+), 1 deletion(-) diff --git a/.github/workflows/examples.yml b/.github/workflows/examples.yml index c34ec5b014a..d1ed9b11493 100644 --- a/.github/workflows/examples.yml +++ b/.github/workflows/examples.yml @@ -82,7 +82,7 @@ jobs: - name: Install reflex 0.8.23 into a fresh venv run: | uv venv --python 3.12 "$RUNNER_TEMP/reflex-0.8.23" - uv pip install --python "$RUNNER_TEMP/reflex-0.8.23/bin/python" "reflex==0.8.23" "plotly>=6.0,<7.0" + uv pip install --python "$RUNNER_TEMP/reflex-0.8.23/bin/python" "reflex==0.8.23" "plotly>=6.0,<7.0" greenlet echo "$RUNNER_TEMP/reflex-0.8.23/bin" >> "$GITHUB_PATH" - name: Show the reflex under test working-directory: examples/playground diff --git a/reflex/state.py b/reflex/state.py index 08028c2f3ed..e714b7f4daa 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -2501,7 +2501,13 @@ async def hydrate_and_load( for state_name, changes in state_delta.items(): correction_delta.setdefault(state_name, {}).update(changes) changed_vars = [] + mutated_storage_vars = {(id(state), name) for state, name in mutated_vars} for var_state, var_name in applied: + # An override may call super().get_delta() and then assign a + # correction. Its returned delta still contains the value from + # before that assignment; the captured write is authoritative. + if (id(var_state), var_name) in mutated_storage_vars: + continue key = var_name + FIELD_MARKER subdelta = correction_delta.get(var_state.get_full_name(), {}) browser_value = var_state.get_value(var_name) diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index 19152e77847..7a84b453e6d 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -2204,6 +2204,62 @@ async def value_type(self) -> str: assert storage_state.value is True +@pytest.mark.parametrize( + "processor_state_manager", ["in_process", "redis"], indirect=True +) +async def test_hydrate_preserves_storage_assignment_after_get_delta_super( + wired_app: App, + real_base_state_processor: BaseStateEventProcessor, + emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], + token: str, +): + """Do not restore an old delta value over a storage assignment in an override. + + Args: + wired_app: The app wired to the state manager. + real_base_state_processor: The event processor. + emitted_deltas: Captured deltas. + token: The client token. + """ + get_delta_calls = 0 + reconciled = False + + class StorageState(State): + value: str = rx.LocalStorage("") + + @_override_base_method + def get_delta(self) -> Delta: + nonlocal get_delta_calls, reconciled + delta = super().get_delta() + get_delta_calls += 1 + if get_delta_calls == 1 and not reconciled and self.value == "stale": + self.value = "fresh" + reconciled = True + return delta + + wired_app.add_page(lambda: rx.text(StorageState.value), route="/") + wired_app._compile_page("index") + boot_name = Event.from_event_type(State.hydrate_and_load())[0].name # pyright: ignore[reportCallIssue] + payload = {"vars": {f"{StorageState.get_full_name()}.value{FIELD_MARKER}": "stale"}} + + async with real_base_state_processor as processor: + await ( + await processor.enqueue(token, _boot_event(boot_name, payload)) + ).wait_all() + + value_key = "value" + FIELD_MARKER + storage_deltas = [ + delta[StorageState.get_full_name()][value_key] + for _, delta in emitted_deltas + if value_key in delta.get(StorageState.get_full_name(), {}) + ] + assert reconciled + assert storage_deltas[-1] == "fresh" + async with _read_back(real_base_state_processor, token) as root: + storage_state = await root.get_state(StorageState) + assert storage_state.value == "fresh" + + async def test_hydrate_probes_nested_storage_states_once( wired_app: App, real_base_state_processor: BaseStateEventProcessor, From cc4fb274ab3933bf6a303711da621caf2326cad9 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 13:37:20 +0530 Subject: [PATCH 13/14] Simplify client storage hydration --- reflex/istate/delta.py | 60 +------ reflex/state.py | 149 +----------------- .../processor/test_base_state_processor.py | 91 ++++------- 3 files changed, 40 insertions(+), 260 deletions(-) diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index f140402c2dc..d312f52e99b 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -6,7 +6,7 @@ import contextlib import inspect import time -from collections.abc import Iterable, Iterator, Mapping +from collections.abc import Iterator, Mapping from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Final, NamedTuple @@ -41,13 +41,6 @@ "_record_delta_values", default=True ) -# During the client-storage reconciliation probe, dirty storage vars must reach -# get_delta overrides without invalidating their computed dependents. Expired -# and always-dirty computed vars are still processed by build_delta. -_suppressed_delta_dependency_sources: ContextVar[frozenset[tuple[int, str]]] = ( - ContextVar("_suppressed_delta_dependency_sources", default=frozenset()) -) - class _DeltaRecord(NamedTuple): """An uncached var value that counts as sent once the delta delivers it.""" @@ -90,49 +83,6 @@ def _suppress_delta_recording() -> Iterator[None]: _record_delta_values.reset(token) -@contextlib.contextmanager -def _suppress_computed_var_dependency_invalidation( - sources: frozenset[tuple[int, str]], -) -> Iterator[None]: - """Suppress computed-var invalidation for selected state vars. - - Args: - sources: State instance ids and var names whose dependents should not be - invalidated during the traversal. - - Yields: - None, with computed-var dependency invalidation suppressed. - """ - token = _suppressed_delta_dependency_sources.set(sources) - try: - yield - finally: - _suppressed_delta_dependency_sources.reset(token) - - -def _get_unsuppressed_var_names( - state: BaseState, - var_names: Iterable[str] | None, - suppressed: frozenset[tuple[int, str]], -) -> Iterable[str] | None: - """Exclude selected source vars from computed-var invalidation. - - Args: - state: The state whose dependencies are being invalidated. - var_names: The changed var names, or None to use all dirty vars. - suppressed: The state vars excluded from dependency invalidation. - - Returns: - The var names that should invalidate computed dependencies. - """ - excluded = {name for state_id, name in suppressed if state_id == id(state)} - return ( - state.dirty_vars - excluded - if var_names is None - else (name for name in var_names if name not in excluded) - ) - - def _commit_delta_records(pending: list[_DeltaRecord], delta: Delta) -> None: """Record what a delivered delta leaves the client holding for each value. @@ -277,13 +227,7 @@ def build_delta(state: BaseState) -> Delta: """ delta = {} - suppressed_sources = _suppressed_delta_dependency_sources.get() - if suppressed_sources: - state._mark_dirty_computed_vars( - _get_unsuppressed_var_names(state, state.dirty_vars, suppressed_sources) - ) - else: - state._mark_dirty_computed_vars() + state._mark_dirty_computed_vars() delta_vars = state.dirty_vars & state._frontend_var_names always_dirty_computed_vars = state._always_dirty_computed_vars diff --git a/reflex/state.py b/reflex/state.py index e714b7f4daa..667729fce59 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -12,9 +12,7 @@ import pickle import re import sys -import threading from collections.abc import Callable, Iterable, Iterator, Sequence -from contextvars import ContextVar from datetime import timedelta from hashlib import md5 from types import FunctionType @@ -90,7 +88,6 @@ Delta, DeltaMapping, _resolve_delta, - _suppress_computed_var_dependency_invalidation, build_delta, clean_state, resolve_delta, @@ -101,49 +98,6 @@ from reflex.utils import console, format, types from reflex.utils.exec import is_testing_env -_dirty_var_collector: ContextVar[set[tuple[BaseState, str]] | None] = ContextVar( - "_dirty_var_collector", default=None -) -_dirty_var_collection_depth = 0 -_dirty_var_collection_lock = threading.Lock() - - -@contextlib.contextmanager -def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: - """Collect vars dirtied by assignments in the current context. - - Yields: - The set of state vars dirtied while the context is active. - """ - global _dirty_var_collection_depth - dirty_vars: set[tuple[BaseState, str]] = set() - with _dirty_var_collection_lock: - _dirty_var_collection_depth += 1 - token = _dirty_var_collector.set(dirty_vars) - try: - yield dirty_vars - finally: - _dirty_var_collector.reset(token) - with _dirty_var_collection_lock: - _dirty_var_collection_depth -= 1 - - -def _state_ancestors(state: BaseState) -> Iterator[BaseState]: - """Yield the parents of a state, from nearest to root.""" - parent = state.parent_state - while parent is not None: - yield parent - parent = parent.parent_state - - -def _dirty_delta_roots(state: BaseState) -> Iterator[BaseState]: - """Yield the highest dirty states whose deltas include their dirty descendants.""" - if state.dirty_vars: - yield state - return - for substate_name in state.dirty_substates: - yield from _dirty_delta_roots(state.substates[substate_name]) - @functools.cache def _stale_pickle_keys(cls: type[BaseState]) -> frozenset[str]: @@ -1989,13 +1943,6 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: Args: var_names: The vars of this state that changed; all its dirty vars if omitted. """ - if _dirty_var_collection_depth: - collector = _dirty_var_collector.get() - if collector is not None: - var_names = tuple(self.dirty_vars if var_names is None else var_names) - collector.update( - (self, name) for name in var_names if name not in self.computed_vars - ) self._mark_ancestors_dirty() self._mark_dirty_computed_vars(var_names) @@ -2430,12 +2377,9 @@ async def hydrate_and_load( storage reset, the browser-provided client storage values and the state snapshot under one state lock avoids separate load/persist cycles of the state tree. The snapshot does not pass through - ``get_delta`` and may not write browser storage, so this event's own - delta carries the browser-provided values again, as - ``update_vars_internal``'s does: ``get_delta`` overrides see them - there, and the browser stores what they return. A page with on_load - handlers then gets ``on_load_internal``, which only locks its leaf - substate. + ``get_delta``; browser-provided values are included in the snapshot + itself. A page with on_load handlers then gets ``on_load_internal``, + which only locks its leaf substate. Args: vars: Client storage vars set in the browser, keyed by fully @@ -2452,7 +2396,8 @@ async def hydrate_and_load( from reflex_base.event.context import EventContext self._reset_client_storage() - applied = await _apply_client_storage_vars(self, vars) if vars else [] + if vars: + await _apply_client_storage_vars(self, vars) self._clean() # The snapshot must carry is_hydrated=False: the frontend skips # writing client storage for a delta that is not yet hydrated, and @@ -2467,81 +2412,8 @@ async def hydrate_and_load( for state_name, changes in (await self._get_resolved_delta()).items(): delta.setdefault(state_name, {}).update(changes) await ctx.emit_delta(delta=delta) - # Follow-up corrections must be allowed to write browser storage. + # Allow subsequent events to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) - if applied: - applied_states = {id(state): state for state, _ in applied} - storage_states = { - state_id: state - for state_id, state in applied_states.items() - if not any( - id(ancestor) in applied_states - for ancestor in _state_ancestors(state) - ) - } - probe_root = any( - type(ancestor).get_delta is not BaseState.get_delta - for state in storage_states.values() - for ancestor in _state_ancestors(state) - ) - # Parent overrides wrap child deltas, so keep their root traversal. - if probe_root: - storage_states = {id(self): self} - for var_state, var_name in applied: - var_state.dirty_vars.add(var_name) - var_state._mark_ancestors_dirty() - suppressed_sources = frozenset((id(state), name) for state, name in applied) - with ( - _collect_dirty_vars() as mutated_vars, - _suppress_computed_var_dependency_invalidation(suppressed_sources), - ): - correction_delta: Delta = {} - for state in storage_states.values(): - state_delta = await state._get_resolved_delta() - for state_name, changes in state_delta.items(): - correction_delta.setdefault(state_name, {}).update(changes) - changed_vars = [] - mutated_storage_vars = {(id(state), name) for state, name in mutated_vars} - for var_state, var_name in applied: - # An override may call super().get_delta() and then assign a - # correction. Its returned delta still contains the value from - # before that assignment; the captured write is authoritative. - if (id(var_state), var_name) in mutated_storage_vars: - continue - key = var_name + FIELD_MARKER - subdelta = correction_delta.get(var_state.get_full_name(), {}) - browser_value = var_state.get_value(var_name) - overridden = key in subdelta and _serialize_var( - subdelta[key] - ) != _serialize_var(browser_value) - if overridden: - setattr(var_state, var_name, subdelta[key]) - changed_vars.append((var_state, var_name)) - if changed_vars or mutated_vars: - # Discard the probe's dirtiness before producing a full-tree - # delta. This makes dependencies on sibling states visible - # and avoids evaluating interval vars in the probe again. - self._clean() - for var_state, var_name in [*changed_vars, *mutated_vars]: - var_state.dirty_vars.add(var_name) - var_state._mark_dirty((var_name,)) - correction_states = list(_dirty_delta_roots(self)) - if any( - type(ancestor).get_delta is not BaseState.get_delta - for state in correction_states - for ancestor in _state_ancestors(state) - ): - correction_states = [self] - corrected_delta: Delta = {} - for state in correction_states: - state_delta = await state._get_resolved_delta() - for state_name, changes in state_delta.items(): - corrected_delta.setdefault(state_name, {}).update(changes) - for state_name, changes in corrected_delta.items(): - correction_delta.setdefault(state_name, {}).update(changes) - if ctx.emit_delta_impl is not None and correction_delta: - await ctx.emit_delta(delta=correction_delta) - self._clean() if not RegistrationContext.get().app.get_load_events(self.rx_router_url.path): self.is_hydrated = True return None @@ -2702,19 +2574,14 @@ async def _diff_against_initial_state( return diff -async def _apply_client_storage_vars( - state: BaseState, vars: dict[str, Any] -) -> list[tuple[BaseState, str]]: +async def _apply_client_storage_vars(state: BaseState, vars: dict[str, Any]) -> None: """Apply browser-provided client storage values to the states that own them. Args: state: Any state in the tree; used to reach the owning substates. vars: Fully qualified var names mapped to their browser values. - Returns: - Each var that was set, with the state that owns it. """ - applied: list[tuple[BaseState, str]] = [] for var, value in vars.items(): state_name, _, var_name = var.rpartition(".") var_name = var_name.removesuffix(FIELD_MARKER) @@ -2722,8 +2589,6 @@ async def _apply_client_storage_vars( if var_state_cls._is_client_storage(var_name): var_state = await state.get_state(var_state_cls) setattr(var_state, var_name, value) - applied.append((var_state, var_name)) - return applied def _load_events_for_page( diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index 7a84b453e6d..d6f0a7311cb 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -1703,15 +1703,13 @@ def on_load_handler(self): ) await future.wait_all() - # Snapshot (not hydrated, browser cookie applied), the browser cookie - # through get_delta (as update_vars_internal sends it), the on_load chain, + # Snapshot (not hydrated, browser cookie applied), the on_load chain, # hydrated. snapshot = emitted_deltas[0][1] assert snapshot[state_name][hydrated_key] is False assert snapshot[CookieState.get_full_name()]["flavor" + FIELD_MARKER] == "chocolate" assert snapshot[CookieState.get_full_name()]["loads" + FIELD_MARKER] == 0 assert [d for _, d in emitted_deltas[1:]] == [ - {CookieState.get_full_name(): {"flavor" + FIELD_MARKER: "chocolate"}}, {state_name: {hydrated_key: False}}, {CookieState.get_full_name(): {"loads" + FIELD_MARKER: 1}}, {state_name: {hydrated_key: True}}, @@ -2013,7 +2011,7 @@ def load(self): @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_passes_client_storage_through_get_delta( +async def test_hydrate_computes_storage_dependents_once( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], @@ -2022,12 +2020,10 @@ async def test_hydrate_passes_client_storage_through_get_delta( with_hashes: bool, client_value: str, ): - """A get_delta override reconciles browser storage without redundant recomputation. + """Hydration does not run a second get_delta pass for client storage. - Overrides reconcile client storage there (reflex-enterprise checks its - OIDC token hash), so the boot values must reach them as - update_vars_internal's did, in a delta the browser may persist, without - writing back the defaults of values the browser does not hold. + A browser-provided value is included in the initial snapshot, without a + second delta that reevaluates dependent computed vars. Args: wired_app: The app wired to the state manager. @@ -2036,7 +2032,7 @@ async def test_hydrate_passes_client_storage_through_get_delta( token: The client token. with_load: Whether the page has an on-load handler. with_hashes: Whether this is the first boot or a reconnect. - client_value: The browser-provided value to reconcile. + client_value: The browser-provided value. """ key = "token_hash" + FIELD_MARKER seen: list[str] = [] @@ -2045,22 +2041,18 @@ async def test_hydrate_passes_client_storage_through_get_delta( class ReconcileState(State): token_hash: str = rx.LocalStorage("") unset: str = rx.Cookie("unset-default") - status: str = "initial" @_override_base_method def get_delta(self) -> Delta: - """Record the token hash passing through and replace a stale one. + """Record the token hash if the override is called. Returns: - The delta, with a stale token hash reconciled. + The parent delta. """ delta = super().get_delta() subdelta = delta.get(self.get_full_name(), {}) if (value := subdelta.get(key)) is not None: seen.append(value) - if value == "stale": - subdelta[key] = "fresh" - self.status = "reconciled" return delta @event @@ -2076,26 +2068,15 @@ async def token_hash_copy(self) -> str: computed_values.append(storage_state.token_hash) return storage_state.token_hash - @rx.var(cache=True, auto_deps=False) - async def status_copy(self) -> str: - storage_state = await self.get_state(ReconcileState) - return storage_state.status - DependentState.computed_vars["token_hash_copy"].add_dependency( DependentState, ReconcileState.token_hash, # pyright: ignore[reportArgumentType] ) - DependentState.computed_vars["status_copy"].add_dependency( - DependentState, - ReconcileState.status, # pyright: ignore[reportArgumentType] - ) - wired_app.add_page( lambda: rx.text( ReconcileState.token_hash, ReconcileState.unset, DependentState.token_hash_copy, - DependentState.status_copy, ), route="/", on_load=ReconcileState.load if with_load else None, @@ -2113,44 +2094,34 @@ async def status_copy(self) -> str: await processor.enqueue(token, _boot_event(boot_name, payload)) ).wait_all() - assert seen == (["stale", "fresh"] if client_value == "stale" else ["fresh"]) + assert seen == [] hydrated_key = CompileVars.IS_HYDRATED + FIELD_MARKER - assert emitted_deltas[0][1][State.get_full_name()][hydrated_key] is False + snapshot = emitted_deltas[0][1] + assert snapshot[State.get_full_name()][hydrated_key] is False + assert snapshot[name][key] == client_value persisted = [ delta[name] for _, delta in emitted_deltas[1:] if name in delta and delta.get(State.get_full_name(), {}).get(hydrated_key) is not False ] - assert any(subdelta.get(key) == "fresh" for subdelta in persisted) + assert not any(subdelta.get(key) == "fresh" for subdelta in persisted) assert not any("unset" + FIELD_MARKER in subdelta for subdelta in persisted) dependent_key = "token_hash_copy" + FIELD_MARKER - assert any( - delta.get(DependentState.get_full_name(), {}).get(dependent_key) == "fresh" - for _, delta in emitted_deltas - ) - assert computed_values == ( - ["stale", "fresh"] if client_value == "stale" else ["fresh"] - ) - status_key = "status_copy" + FIELD_MARKER - if client_value == "stale": - assert any( - delta.get(DependentState.get_full_name(), {}).get(status_key) - == "reconciled" - for _, delta in emitted_deltas - ) + assert snapshot[DependentState.get_full_name()][dependent_key] == client_value + assert computed_values == [client_value] @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_reconciles_json_distinct_storage_values( +async def test_hydrate_preserves_json_distinct_storage_values( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """Hydration treats JSON-distinct values as changes even when Python equates them. + """Hydration does not pass client storage through get_delta overrides. Args: wired_app: The app wired to the state manager. @@ -2195,25 +2166,25 @@ async def value_type(self) -> str: ).wait_all() value_type_key = "value_type" + FIELD_MARKER - assert any( - delta.get(DependentState.get_full_name(), {}).get(value_type_key) == "bool" - for _, delta in emitted_deltas + assert ( + emitted_deltas[0][1][StorageState.get_full_name()]["value" + FIELD_MARKER] == 1 ) + assert emitted_deltas[0][1][DependentState.get_full_name()][value_type_key] == "int" async with _read_back(real_base_state_processor, token) as root: storage_state = await root.get_state(StorageState) - assert storage_state.value is True + assert storage_state.value == 1 @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_preserves_storage_assignment_after_get_delta_super( +async def test_hydrate_does_not_reconcile_storage_after_snapshot( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """Do not restore an old delta value over a storage assignment in an override. + """Hydration does not invoke a storage reconciliation override after snapshotting. Args: wired_app: The app wired to the state manager. @@ -2253,11 +2224,11 @@ def get_delta(self) -> Delta: for _, delta in emitted_deltas if value_key in delta.get(StorageState.get_full_name(), {}) ] - assert reconciled - assert storage_deltas[-1] == "fresh" + assert not reconciled + assert storage_deltas[-1] == "stale" async with _read_back(real_base_state_processor, token) as root: storage_state = await root.get_state(StorageState) - assert storage_state.value == "fresh" + assert storage_state.value == "stale" async def test_hydrate_probes_nested_storage_states_once( @@ -2265,7 +2236,7 @@ async def test_hydrate_probes_nested_storage_states_once( real_base_state_processor: BaseStateEventProcessor, token: str, ): - """A parent probe already visits child states that also own storage vars. + """Hydration does not probe child get_delta overrides after the snapshot. Args: wired_app: The app wired to the state manager. @@ -2309,7 +2280,7 @@ def get_delta(self) -> Delta: await processor.enqueue(token, _boot_event(boot_name, payload)) ).wait_all() - assert child_storage_probes == 1 + assert child_storage_probes == 0 async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( @@ -2318,7 +2289,7 @@ async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """A stale storage correction does not reevaluate unrelated state branches. + """Hydration does not run an extra correction pass for stale storage. Args: wired_app: The app wired to the state manager. @@ -2378,10 +2349,10 @@ async def hydrate(value: str) -> int: stale_computations = await hydrate("stale") assert stale_computations == fresh_computations - assert any( + assert not any( delta.get(StorageState.get_full_name(), {}).get("value" + FIELD_MARKER) == "fresh" - for _, delta in emitted_deltas + for _, delta in emitted_deltas[1:] ) From 0a8098d5ace516ea2b3441ec1d58d5ae58d59861 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 13:50:02 +0530 Subject: [PATCH 14/14] Revert "Simplify client storage hydration" This reverts commit cc4fb274ab3933bf6a303711da621caf2326cad9. --- reflex/istate/delta.py | 60 ++++++- reflex/state.py | 149 +++++++++++++++++- .../processor/test_base_state_processor.py | 91 +++++++---- 3 files changed, 260 insertions(+), 40 deletions(-) diff --git a/reflex/istate/delta.py b/reflex/istate/delta.py index d312f52e99b..f140402c2dc 100644 --- a/reflex/istate/delta.py +++ b/reflex/istate/delta.py @@ -6,7 +6,7 @@ import contextlib import inspect import time -from collections.abc import Iterator, Mapping +from collections.abc import Iterable, Iterator, Mapping from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Final, NamedTuple @@ -41,6 +41,13 @@ "_record_delta_values", default=True ) +# During the client-storage reconciliation probe, dirty storage vars must reach +# get_delta overrides without invalidating their computed dependents. Expired +# and always-dirty computed vars are still processed by build_delta. +_suppressed_delta_dependency_sources: ContextVar[frozenset[tuple[int, str]]] = ( + ContextVar("_suppressed_delta_dependency_sources", default=frozenset()) +) + class _DeltaRecord(NamedTuple): """An uncached var value that counts as sent once the delta delivers it.""" @@ -83,6 +90,49 @@ def _suppress_delta_recording() -> Iterator[None]: _record_delta_values.reset(token) +@contextlib.contextmanager +def _suppress_computed_var_dependency_invalidation( + sources: frozenset[tuple[int, str]], +) -> Iterator[None]: + """Suppress computed-var invalidation for selected state vars. + + Args: + sources: State instance ids and var names whose dependents should not be + invalidated during the traversal. + + Yields: + None, with computed-var dependency invalidation suppressed. + """ + token = _suppressed_delta_dependency_sources.set(sources) + try: + yield + finally: + _suppressed_delta_dependency_sources.reset(token) + + +def _get_unsuppressed_var_names( + state: BaseState, + var_names: Iterable[str] | None, + suppressed: frozenset[tuple[int, str]], +) -> Iterable[str] | None: + """Exclude selected source vars from computed-var invalidation. + + Args: + state: The state whose dependencies are being invalidated. + var_names: The changed var names, or None to use all dirty vars. + suppressed: The state vars excluded from dependency invalidation. + + Returns: + The var names that should invalidate computed dependencies. + """ + excluded = {name for state_id, name in suppressed if state_id == id(state)} + return ( + state.dirty_vars - excluded + if var_names is None + else (name for name in var_names if name not in excluded) + ) + + def _commit_delta_records(pending: list[_DeltaRecord], delta: Delta) -> None: """Record what a delivered delta leaves the client holding for each value. @@ -227,7 +277,13 @@ def build_delta(state: BaseState) -> Delta: """ delta = {} - state._mark_dirty_computed_vars() + suppressed_sources = _suppressed_delta_dependency_sources.get() + if suppressed_sources: + state._mark_dirty_computed_vars( + _get_unsuppressed_var_names(state, state.dirty_vars, suppressed_sources) + ) + else: + state._mark_dirty_computed_vars() delta_vars = state.dirty_vars & state._frontend_var_names always_dirty_computed_vars = state._always_dirty_computed_vars diff --git a/reflex/state.py b/reflex/state.py index 667729fce59..e714b7f4daa 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -12,7 +12,9 @@ import pickle import re import sys +import threading from collections.abc import Callable, Iterable, Iterator, Sequence +from contextvars import ContextVar from datetime import timedelta from hashlib import md5 from types import FunctionType @@ -88,6 +90,7 @@ Delta, DeltaMapping, _resolve_delta, + _suppress_computed_var_dependency_invalidation, build_delta, clean_state, resolve_delta, @@ -98,6 +101,49 @@ from reflex.utils import console, format, types from reflex.utils.exec import is_testing_env +_dirty_var_collector: ContextVar[set[tuple[BaseState, str]] | None] = ContextVar( + "_dirty_var_collector", default=None +) +_dirty_var_collection_depth = 0 +_dirty_var_collection_lock = threading.Lock() + + +@contextlib.contextmanager +def _collect_dirty_vars() -> Iterator[set[tuple[BaseState, str]]]: + """Collect vars dirtied by assignments in the current context. + + Yields: + The set of state vars dirtied while the context is active. + """ + global _dirty_var_collection_depth + dirty_vars: set[tuple[BaseState, str]] = set() + with _dirty_var_collection_lock: + _dirty_var_collection_depth += 1 + token = _dirty_var_collector.set(dirty_vars) + try: + yield dirty_vars + finally: + _dirty_var_collector.reset(token) + with _dirty_var_collection_lock: + _dirty_var_collection_depth -= 1 + + +def _state_ancestors(state: BaseState) -> Iterator[BaseState]: + """Yield the parents of a state, from nearest to root.""" + parent = state.parent_state + while parent is not None: + yield parent + parent = parent.parent_state + + +def _dirty_delta_roots(state: BaseState) -> Iterator[BaseState]: + """Yield the highest dirty states whose deltas include their dirty descendants.""" + if state.dirty_vars: + yield state + return + for substate_name in state.dirty_substates: + yield from _dirty_delta_roots(state.substates[substate_name]) + @functools.cache def _stale_pickle_keys(cls: type[BaseState]) -> frozenset[str]: @@ -1943,6 +1989,13 @@ def _mark_dirty(self, var_names: Iterable[str] | None = None) -> None: Args: var_names: The vars of this state that changed; all its dirty vars if omitted. """ + if _dirty_var_collection_depth: + collector = _dirty_var_collector.get() + if collector is not None: + var_names = tuple(self.dirty_vars if var_names is None else var_names) + collector.update( + (self, name) for name in var_names if name not in self.computed_vars + ) self._mark_ancestors_dirty() self._mark_dirty_computed_vars(var_names) @@ -2377,9 +2430,12 @@ async def hydrate_and_load( storage reset, the browser-provided client storage values and the state snapshot under one state lock avoids separate load/persist cycles of the state tree. The snapshot does not pass through - ``get_delta``; browser-provided values are included in the snapshot - itself. A page with on_load handlers then gets ``on_load_internal``, - which only locks its leaf substate. + ``get_delta`` and may not write browser storage, so this event's own + delta carries the browser-provided values again, as + ``update_vars_internal``'s does: ``get_delta`` overrides see them + there, and the browser stores what they return. A page with on_load + handlers then gets ``on_load_internal``, which only locks its leaf + substate. Args: vars: Client storage vars set in the browser, keyed by fully @@ -2396,8 +2452,7 @@ async def hydrate_and_load( from reflex_base.event.context import EventContext self._reset_client_storage() - if vars: - await _apply_client_storage_vars(self, vars) + applied = await _apply_client_storage_vars(self, vars) if vars else [] self._clean() # The snapshot must carry is_hydrated=False: the frontend skips # writing client storage for a delta that is not yet hydrated, and @@ -2412,8 +2467,81 @@ async def hydrate_and_load( for state_name, changes in (await self._get_resolved_delta()).items(): delta.setdefault(state_name, {}).update(changes) await ctx.emit_delta(delta=delta) - # Allow subsequent events to write browser storage. + # Follow-up corrections must be allowed to write browser storage. self.dirty_vars.discard(constants.CompileVars.IS_HYDRATED) + if applied: + applied_states = {id(state): state for state, _ in applied} + storage_states = { + state_id: state + for state_id, state in applied_states.items() + if not any( + id(ancestor) in applied_states + for ancestor in _state_ancestors(state) + ) + } + probe_root = any( + type(ancestor).get_delta is not BaseState.get_delta + for state in storage_states.values() + for ancestor in _state_ancestors(state) + ) + # Parent overrides wrap child deltas, so keep their root traversal. + if probe_root: + storage_states = {id(self): self} + for var_state, var_name in applied: + var_state.dirty_vars.add(var_name) + var_state._mark_ancestors_dirty() + suppressed_sources = frozenset((id(state), name) for state, name in applied) + with ( + _collect_dirty_vars() as mutated_vars, + _suppress_computed_var_dependency_invalidation(suppressed_sources), + ): + correction_delta: Delta = {} + for state in storage_states.values(): + state_delta = await state._get_resolved_delta() + for state_name, changes in state_delta.items(): + correction_delta.setdefault(state_name, {}).update(changes) + changed_vars = [] + mutated_storage_vars = {(id(state), name) for state, name in mutated_vars} + for var_state, var_name in applied: + # An override may call super().get_delta() and then assign a + # correction. Its returned delta still contains the value from + # before that assignment; the captured write is authoritative. + if (id(var_state), var_name) in mutated_storage_vars: + continue + key = var_name + FIELD_MARKER + subdelta = correction_delta.get(var_state.get_full_name(), {}) + browser_value = var_state.get_value(var_name) + overridden = key in subdelta and _serialize_var( + subdelta[key] + ) != _serialize_var(browser_value) + if overridden: + setattr(var_state, var_name, subdelta[key]) + changed_vars.append((var_state, var_name)) + if changed_vars or mutated_vars: + # Discard the probe's dirtiness before producing a full-tree + # delta. This makes dependencies on sibling states visible + # and avoids evaluating interval vars in the probe again. + self._clean() + for var_state, var_name in [*changed_vars, *mutated_vars]: + var_state.dirty_vars.add(var_name) + var_state._mark_dirty((var_name,)) + correction_states = list(_dirty_delta_roots(self)) + if any( + type(ancestor).get_delta is not BaseState.get_delta + for state in correction_states + for ancestor in _state_ancestors(state) + ): + correction_states = [self] + corrected_delta: Delta = {} + for state in correction_states: + state_delta = await state._get_resolved_delta() + for state_name, changes in state_delta.items(): + corrected_delta.setdefault(state_name, {}).update(changes) + for state_name, changes in corrected_delta.items(): + correction_delta.setdefault(state_name, {}).update(changes) + if ctx.emit_delta_impl is not None and correction_delta: + await ctx.emit_delta(delta=correction_delta) + self._clean() if not RegistrationContext.get().app.get_load_events(self.rx_router_url.path): self.is_hydrated = True return None @@ -2574,14 +2702,19 @@ async def _diff_against_initial_state( return diff -async def _apply_client_storage_vars(state: BaseState, vars: dict[str, Any]) -> None: +async def _apply_client_storage_vars( + state: BaseState, vars: dict[str, Any] +) -> list[tuple[BaseState, str]]: """Apply browser-provided client storage values to the states that own them. Args: state: Any state in the tree; used to reach the owning substates. vars: Fully qualified var names mapped to their browser values. + Returns: + Each var that was set, with the state that owns it. """ + applied: list[tuple[BaseState, str]] = [] for var, value in vars.items(): state_name, _, var_name = var.rpartition(".") var_name = var_name.removesuffix(FIELD_MARKER) @@ -2589,6 +2722,8 @@ async def _apply_client_storage_vars(state: BaseState, vars: dict[str, Any]) -> if var_state_cls._is_client_storage(var_name): var_state = await state.get_state(var_state_cls) setattr(var_state, var_name, value) + applied.append((var_state, var_name)) + return applied def _load_events_for_page( diff --git a/tests/units/reflex_base/event/processor/test_base_state_processor.py b/tests/units/reflex_base/event/processor/test_base_state_processor.py index d6f0a7311cb..7a84b453e6d 100644 --- a/tests/units/reflex_base/event/processor/test_base_state_processor.py +++ b/tests/units/reflex_base/event/processor/test_base_state_processor.py @@ -1703,13 +1703,15 @@ def on_load_handler(self): ) await future.wait_all() - # Snapshot (not hydrated, browser cookie applied), the on_load chain, + # Snapshot (not hydrated, browser cookie applied), the browser cookie + # through get_delta (as update_vars_internal sends it), the on_load chain, # hydrated. snapshot = emitted_deltas[0][1] assert snapshot[state_name][hydrated_key] is False assert snapshot[CookieState.get_full_name()]["flavor" + FIELD_MARKER] == "chocolate" assert snapshot[CookieState.get_full_name()]["loads" + FIELD_MARKER] == 0 assert [d for _, d in emitted_deltas[1:]] == [ + {CookieState.get_full_name(): {"flavor" + FIELD_MARKER: "chocolate"}}, {state_name: {hydrated_key: False}}, {CookieState.get_full_name(): {"loads" + FIELD_MARKER: 1}}, {state_name: {hydrated_key: True}}, @@ -2011,7 +2013,7 @@ def load(self): @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_computes_storage_dependents_once( +async def test_hydrate_passes_client_storage_through_get_delta( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], @@ -2020,10 +2022,12 @@ async def test_hydrate_computes_storage_dependents_once( with_hashes: bool, client_value: str, ): - """Hydration does not run a second get_delta pass for client storage. + """A get_delta override reconciles browser storage without redundant recomputation. - A browser-provided value is included in the initial snapshot, without a - second delta that reevaluates dependent computed vars. + Overrides reconcile client storage there (reflex-enterprise checks its + OIDC token hash), so the boot values must reach them as + update_vars_internal's did, in a delta the browser may persist, without + writing back the defaults of values the browser does not hold. Args: wired_app: The app wired to the state manager. @@ -2032,7 +2036,7 @@ async def test_hydrate_computes_storage_dependents_once( token: The client token. with_load: Whether the page has an on-load handler. with_hashes: Whether this is the first boot or a reconnect. - client_value: The browser-provided value. + client_value: The browser-provided value to reconcile. """ key = "token_hash" + FIELD_MARKER seen: list[str] = [] @@ -2041,18 +2045,22 @@ async def test_hydrate_computes_storage_dependents_once( class ReconcileState(State): token_hash: str = rx.LocalStorage("") unset: str = rx.Cookie("unset-default") + status: str = "initial" @_override_base_method def get_delta(self) -> Delta: - """Record the token hash if the override is called. + """Record the token hash passing through and replace a stale one. Returns: - The parent delta. + The delta, with a stale token hash reconciled. """ delta = super().get_delta() subdelta = delta.get(self.get_full_name(), {}) if (value := subdelta.get(key)) is not None: seen.append(value) + if value == "stale": + subdelta[key] = "fresh" + self.status = "reconciled" return delta @event @@ -2068,15 +2076,26 @@ async def token_hash_copy(self) -> str: computed_values.append(storage_state.token_hash) return storage_state.token_hash + @rx.var(cache=True, auto_deps=False) + async def status_copy(self) -> str: + storage_state = await self.get_state(ReconcileState) + return storage_state.status + DependentState.computed_vars["token_hash_copy"].add_dependency( DependentState, ReconcileState.token_hash, # pyright: ignore[reportArgumentType] ) + DependentState.computed_vars["status_copy"].add_dependency( + DependentState, + ReconcileState.status, # pyright: ignore[reportArgumentType] + ) + wired_app.add_page( lambda: rx.text( ReconcileState.token_hash, ReconcileState.unset, DependentState.token_hash_copy, + DependentState.status_copy, ), route="/", on_load=ReconcileState.load if with_load else None, @@ -2094,34 +2113,44 @@ async def token_hash_copy(self) -> str: await processor.enqueue(token, _boot_event(boot_name, payload)) ).wait_all() - assert seen == [] + assert seen == (["stale", "fresh"] if client_value == "stale" else ["fresh"]) hydrated_key = CompileVars.IS_HYDRATED + FIELD_MARKER - snapshot = emitted_deltas[0][1] - assert snapshot[State.get_full_name()][hydrated_key] is False - assert snapshot[name][key] == client_value + assert emitted_deltas[0][1][State.get_full_name()][hydrated_key] is False persisted = [ delta[name] for _, delta in emitted_deltas[1:] if name in delta and delta.get(State.get_full_name(), {}).get(hydrated_key) is not False ] - assert not any(subdelta.get(key) == "fresh" for subdelta in persisted) + assert any(subdelta.get(key) == "fresh" for subdelta in persisted) assert not any("unset" + FIELD_MARKER in subdelta for subdelta in persisted) dependent_key = "token_hash_copy" + FIELD_MARKER - assert snapshot[DependentState.get_full_name()][dependent_key] == client_value - assert computed_values == [client_value] + assert any( + delta.get(DependentState.get_full_name(), {}).get(dependent_key) == "fresh" + for _, delta in emitted_deltas + ) + assert computed_values == ( + ["stale", "fresh"] if client_value == "stale" else ["fresh"] + ) + status_key = "status_copy" + FIELD_MARKER + if client_value == "stale": + assert any( + delta.get(DependentState.get_full_name(), {}).get(status_key) + == "reconciled" + for _, delta in emitted_deltas + ) @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_preserves_json_distinct_storage_values( +async def test_hydrate_reconciles_json_distinct_storage_values( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """Hydration does not pass client storage through get_delta overrides. + """Hydration treats JSON-distinct values as changes even when Python equates them. Args: wired_app: The app wired to the state manager. @@ -2166,25 +2195,25 @@ async def value_type(self) -> str: ).wait_all() value_type_key = "value_type" + FIELD_MARKER - assert ( - emitted_deltas[0][1][StorageState.get_full_name()]["value" + FIELD_MARKER] == 1 + assert any( + delta.get(DependentState.get_full_name(), {}).get(value_type_key) == "bool" + for _, delta in emitted_deltas ) - assert emitted_deltas[0][1][DependentState.get_full_name()][value_type_key] == "int" async with _read_back(real_base_state_processor, token) as root: storage_state = await root.get_state(StorageState) - assert storage_state.value == 1 + assert storage_state.value is True @pytest.mark.parametrize( "processor_state_manager", ["in_process", "redis"], indirect=True ) -async def test_hydrate_does_not_reconcile_storage_after_snapshot( +async def test_hydrate_preserves_storage_assignment_after_get_delta_super( wired_app: App, real_base_state_processor: BaseStateEventProcessor, emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """Hydration does not invoke a storage reconciliation override after snapshotting. + """Do not restore an old delta value over a storage assignment in an override. Args: wired_app: The app wired to the state manager. @@ -2224,11 +2253,11 @@ def get_delta(self) -> Delta: for _, delta in emitted_deltas if value_key in delta.get(StorageState.get_full_name(), {}) ] - assert not reconciled - assert storage_deltas[-1] == "stale" + assert reconciled + assert storage_deltas[-1] == "fresh" async with _read_back(real_base_state_processor, token) as root: storage_state = await root.get_state(StorageState) - assert storage_state.value == "stale" + assert storage_state.value == "fresh" async def test_hydrate_probes_nested_storage_states_once( @@ -2236,7 +2265,7 @@ async def test_hydrate_probes_nested_storage_states_once( real_base_state_processor: BaseStateEventProcessor, token: str, ): - """Hydration does not probe child get_delta overrides after the snapshot. + """A parent probe already visits child states that also own storage vars. Args: wired_app: The app wired to the state manager. @@ -2280,7 +2309,7 @@ def get_delta(self) -> Delta: await processor.enqueue(token, _boot_event(boot_name, payload)) ).wait_all() - assert child_storage_probes == 0 + assert child_storage_probes == 1 async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( @@ -2289,7 +2318,7 @@ async def test_hydrate_correction_skips_unrelated_uncached_computed_vars( emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]], token: str, ): - """Hydration does not run an extra correction pass for stale storage. + """A stale storage correction does not reevaluate unrelated state branches. Args: wired_app: The app wired to the state manager. @@ -2349,10 +2378,10 @@ async def hydrate(value: str) -> int: stale_computations = await hydrate("stale") assert stale_computations == fresh_computations - assert not any( + assert any( delta.get(StorageState.get_full_name(), {}).get("value" + FIELD_MARKER) == "fresh" - for _, delta in emitted_deltas[1:] + for _, delta in emitted_deltas )