diff --git a/AGENTS.md b/AGENTS.md index 33d278a6e23..0eb7606c063 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -115,6 +115,17 @@ documentation — write it under `docs/` and let the fragment link there. CI requires a fragment for every package whose source the PR touches; the `skip-changelog` label waives it for changes that are genuinely not user-facing. +Which `news/` directory a fragment lands in is decided purely by changed path: +`reflex/**` -> repo-root `news/`; `packages//src/**` -> +`packages//news/`. A PR spanning `reflex/` and `packages/reflex-base/src/` +therefore needs a fragment in both. Paths outside those (tests, `docs/`, CI, +`scripts/`) require none on their own. Check the way CI does, once per affected +package: + +``` +uv run towncrier check --config pyproject.toml --dir --compare-with origin/main +``` + ## Breaking changes and deprecation Reflex has downstream users — don't break them. Provide a fallback path during deprecation. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index efab20e411c..057561d8d28 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -60,7 +60,27 @@ Within the 'test' directory of Reflex you can add to a test file already there o Each PR that changes the source of a published package must add a news fragment describing the change. Fragments are assembled into `CHANGELOG.md` at release time by [towncrier](https://towncrier.readthedocs.io/). -**Where:** add the fragment under the affected package's `news/` directory. For the main `reflex` package, that's the repo-root `news/`. For sub-packages it's `packages//news/`. +**Where: one fragment per affected package, not one per PR.** A PR that touches +two packages needs two fragments, one in each package's own `news/` directory. +CI decides which packages are affected purely from the changed paths: + +| Changed path | Fragment belongs in | +| --- | --- | +| `reflex/**` | repo-root `news/` | +| `packages//src/**` | `packages//news/` | + +So a PR editing both `reflex/state.py` and `packages/reflex-base/src/reflex_base/vars/base.py` +needs a fragment in `news/` **and** one in `packages/reflex-base/news/`. Changes +that touch neither path (tests, `docs/`, CI, `scripts/`) need no fragment on +their own. The `integrations-docs`, `reflex-components-internal` and +`reflex-site-shared` packages are never published, so they are always exempt. + +Run the same check CI runs, once per affected package: + +```bash +uv run towncrier check --config pyproject.toml --dir . --compare-with origin/main +uv run towncrier check --config pyproject.toml --dir packages/reflex-base --compare-with origin/main +``` **Filename:** `..md`, where `` is one of: diff --git a/docs/vars/computed_vars.md b/docs/vars/computed_vars.md index 36a048e8187..551628b01a4 100644 --- a/docs/vars/computed_vars.md +++ b/docs/vars/computed_vars.md @@ -46,6 +46,10 @@ expensive computations, but in some cases it may not update when you expect it t To create a computed var that recomputes on every state update regardless of dependencies, use `@rx.var(cache=False)`. +An uncached var is recomputed for every state update, but the recomputed value is +only sent to the frontend when it differs from the value that was last sent, so a +recomputation that yields the same value does not trigger a re-render. + Previous versions of Reflex had a `@rx.cached_var` decorator, which is now replaced by the `cache` argument of `@rx.var` (which defaults to `True`). diff --git a/news/+uncached-computed-var-delta-dedupe.performance.md b/news/+uncached-computed-var-delta-dedupe.performance.md new file mode 100644 index 00000000000..796a6877f99 --- /dev/null +++ b/news/+uncached-computed-var-delta-dedupe.performance.md @@ -0,0 +1 @@ +`@rx.var(cache=False)` vars now remember what they last sent to the frontend. They are still recomputed on every state update, but the value is only included in the delta when it actually changed, so an uncached var whose value stays the same no longer causes needless network traffic and re-renders. diff --git a/packages/reflex-base/news/+uncached-computed-var-delta-dedupe.performance.md b/packages/reflex-base/news/+uncached-computed-var-delta-dedupe.performance.md new file mode 100644 index 00000000000..d74fac1db3d --- /dev/null +++ b/packages/reflex-base/news/+uncached-computed-var-delta-dedupe.performance.md @@ -0,0 +1 @@ +`ComputedVar` now records a key for the value an uncached (`cache=False`) var last sent to each client, so `BaseState.get_delta` can leave the var out of the delta when a recomputation produces the same value. diff --git a/packages/reflex-base/src/reflex_base/vars/base.py b/packages/reflex-base/src/reflex_base/vars/base.py index 8803960d51c..e98f02a52f8 100644 --- a/packages/reflex-base/src/reflex_base/vars/base.py +++ b/packages/reflex-base/src/reflex_base/vars/base.py @@ -8,6 +8,7 @@ import dataclasses import datetime import functools +import hashlib import inspect import json import logging @@ -25,6 +26,7 @@ Annotated, Any, ClassVar, + Final, Generic, Literal, NoReturn, @@ -53,7 +55,7 @@ VarDependencyError, VarTypeError, ) -from reflex_base.utils.format import format_state_name +from reflex_base.utils.format import format_state_name, json_dumps from reflex_base.utils.imports import ( ImmutableImportDict, ImmutableParsedImportDict, @@ -2249,6 +2251,47 @@ class FakeComputedVarBaseClass(property): __pydantic_run_validation__ = False +# Marker for a value that has no delta key. Compared by identity and never stored +# on a state instance, so it is safe from serialization round trips. +_UNKEYABLE_VALUE: Final = object() + +# Types whose instances are immutable and cheap to compare directly. float is +# deliberately absent: NaN is not equal to itself, so floats are keyed by their +# serialized form instead of comparing equal to nothing forever. +_ATOMIC_DELTA_VALUE_TYPES: Final = frozenset({str, int, bool, type(None)}) + +# Size of the digest used to key non-atomic delta values. +_DELTA_VALUE_DIGEST_SIZE: Final = 16 + + +def _delta_value_key(value: Any) -> Any: + """Get a comparable, alias-free key for a value going into a delta. + + Immutable scalars are keyed by themselves, paired with their type so that + Python-equal but JSON-distinct values (``1`` and ``True``) do not collide. + Everything else is keyed by a digest of its serialized form: that is exactly + what the client receives, so a value without a registered serializer keys by + the ``null`` the client would get, it cannot be invalidated by a later + in-place mutation of the value, and it stays small no matter how big the + value is. + + Args: + value: The value to key. + + Returns: + The key, or ``_UNKEYABLE_VALUE`` if the value cannot be serialized. + """ + value_type = type(value) + if value_type in _ATOMIC_DELTA_VALUE_TYPES: + return (value_type, value) + try: + return hashlib.blake2b( + json_dumps(value).encode(), digest_size=_DELTA_VALUE_DIGEST_SIZE + ).digest() + except Exception: + return _UNKEYABLE_VALUE + + def is_computed_var(obj: Any) -> TypeGuard[ComputedVar]: """Check if the object is a ComputedVar. @@ -2483,6 +2526,51 @@ def _last_updated_attr(self) -> str: """ return f"__last_updated_{self._js_expr}" + @property + def _last_delta_key_attr(self) -> str: + """The attribute used to store the key of the last value sent in a delta. + + Returns: + An attribute name. + """ + return f"__last_delta_{self._js_expr}" + + def _record_delta_value(self, instance: BaseState, value: Any, token: str) -> bool: + """Record the value an uncached var contributes to the delta. + + Uncached vars are recomputed for every delta, but recomputing does not + imply the value changed. Keeping a key for the last value that was sent + to the client allows an unchanged value to be omitted from the delta, + avoiding a needless re-render on the frontend. + + The client token is recorded alongside the key because a single state + instance can serve several clients (linked shared states): a value that + was already sent to one client still has to be sent to the others. + + Args: + instance: The state instance that the computed var is attached to. + value: The freshly computed value. + token: The client token the delta is being produced for. + + Returns: + Whether the value differs from the last recorded value and should + therefore be included in the delta. + """ + attr = self._last_delta_key_attr + key = _delta_value_key(value) + if key is _UNKEYABLE_VALUE: + # The value can never be compared, so it always has to be sent. + with contextlib.suppress(AttributeError): + delattr(instance, attr) + return True + recorded = (token, key) + if getattr(instance, attr, None) == recorded: + return False + setattr(instance, attr, recorded) + # Ensure the recorded value gets serialized to redis. + instance._was_touched = True + return True + def needs_update(self, instance: BaseState) -> bool: """Check if the computed var needs to be updated. diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index 432cdadbb52..f7bcb9519d6 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -112,7 +112,8 @@ async def _patch_state( root_state.dirty_vars.add("router") root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() - await root_state._get_resolved_delta() + # The delta is discarded: it is only resolved to refresh computed vars. + await root_state._get_resolved_delta(record_values=False) yield finally: original_parent_state.substates[state_name] = original_state diff --git a/reflex/state.py b/reflex/state.py index 2f5442b7e82..25824f3a655 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -14,7 +14,7 @@ import re import sys import time -from collections.abc import Callable, Iterator, Mapping, Sequence +from collections.abc import Callable, Coroutine, Iterator, Mapping, Sequence from hashlib import md5 from types import FunctionType from typing import ( @@ -309,6 +309,30 @@ async def _resolve_delta(delta: Delta) -> Delta: return delta +async def _drop_unchanged_delta_value( + cvar: ComputedVar, + instance: BaseState, + value: Coroutine[None, None, Any], + token: str, +) -> Any: + """Await an async uncached computed var, dropping it if the value did not change. + + Args: + cvar: The computed var that produced the coroutine. + instance: The state instance the computed var is attached to. + value: The coroutine returned by the computed var. + token: The client token the delta is being produced for. + + Returns: + The resolved value, or ``_DROP_FROM_DELTA`` when it matches the last + value that was sent to the client. + """ + resolved = await value + if not cvar._record_delta_value(instance, resolved, token): + return _DROP_FROM_DELTA + return resolved + + RETURN = TypeVar("RETURN") PARAMS = ParamSpec("PARAMS") @@ -1954,9 +1978,15 @@ def _dirty_computed_vars( if include_backend or not self.computed_vars[cvar]._backend } - def get_delta(self) -> Delta: + def get_delta(self, record_values: bool = True) -> Delta: """Get the delta for the state. + Args: + record_values: Whether the values of uncached computed vars should be + recorded as sent to the client. Pass False when the delta is + computed for its side effects and then discarded, otherwise the + unsent values would be omitted from the next delta. + Returns: The delta for the state. """ @@ -1973,11 +2003,23 @@ def get_delta(self) -> Delta: self.dirty_vars.intersection(frontend_computed_vars) ) - subdelta: dict[str, Any] = { - prop + FIELD_MARKER: self.get_value(prop) - for prop in delta_vars - if not types.is_backend_base_variable(prop, type(self)) - } + always_dirty_computed_vars = self._always_dirty_computed_vars + # Token of the client this delta is for, used to know which values it has. + token = self.router.session.client_token if always_dirty_computed_vars else "" + subdelta: dict[str, Any] = {} + for prop in delta_vars: + if types.is_backend_base_variable(prop, type(self)): + continue + value = self.get_value(prop) + if record_values and prop in always_dirty_computed_vars: + # Uncached computed vars are recomputed for every delta; only + # send them when the recomputed value actually changed. + cvar = self.computed_vars[prop] + if inspect.iscoroutine(value): + value = _drop_unchanged_delta_value(cvar, self, value, token) + elif not cvar._record_delta_value(self, value, token): + continue + subdelta[prop + FIELD_MARKER] = value if len(subdelta) > 0: delta[self.get_full_name()] = subdelta @@ -1985,18 +2027,22 @@ def get_delta(self) -> Delta: # Recursively find the substate deltas. substates = self.substates for substate in self.dirty_substates.union(self._always_dirty_substates): - delta.update(substates[substate].get_delta()) + delta.update(substates[substate].get_delta(record_values=record_values)) # Return the delta. return delta - async def _get_resolved_delta(self) -> Delta: + async def _get_resolved_delta(self, record_values: bool = True) -> Delta: """Get the delta for the state after resolving all coroutines. + Args: + record_values: Whether the values of uncached computed vars should be + recorded as sent to the client. See `get_delta`. + Returns: The resolved delta for the state. """ - return await _resolve_delta(self.get_delta()) + return await _resolve_delta(self.get_delta(record_values=record_values)) def _mark_dirty(self): """Mark the substate and all parent states as dirty.""" diff --git a/tests/units/test_state.py b/tests/units/test_state.py index 346bfc4e038..5a51e5f80c5 100644 --- a/tests/units/test_state.py +++ b/tests/units/test_state.py @@ -43,7 +43,7 @@ import reflex as rx from reflex.app import App from reflex.environment import environment -from reflex.istate.data import HeaderData, RouterData, _FrozenDictStrStr +from reflex.istate.data import HeaderData, RouterData, SessionData, _FrozenDictStrStr from reflex.istate.manager import StateManager from reflex.istate.manager.disk import StateManagerDisk from reflex.istate.manager.memory import StateManagerMemory @@ -992,13 +992,12 @@ async def test_process_event_substate( ) async with mock_base_state_event_processor as processor: await processor.enqueue(token, event) + # GrandchildState3.computed is uncached, but its value is unchanged since the + # previous delta, so it is not sent again. assert emitted_deltas == [ ( token, - { - GrandchildState.get_full_name(): {"value2" + FIELD_MARKER: "new"}, - GrandchildState3.get_full_name(): {"computed" + FIELD_MARKER: ""}, - }, + {GrandchildState.get_full_name(): {"value2" + FIELD_MARKER: "new"}}, ) ] @@ -1456,9 +1455,8 @@ def comp_v(self) -> int: } cs._clean() assert cs.dirty_vars == set() - assert cs.get_delta() == { - cs.get_name(): {"no_cache_v" + FIELD_MARKER: 0, "dep_v" + FIELD_MARKER: 0} - } + # no_cache_v is recomputed, but the value is unchanged, so it is not resent. + assert cs.get_delta() == {cs.get_name(): {"dep_v" + FIELD_MARKER: 0}} cs._clean() assert cs.dirty_vars == set() cs.v = 1 @@ -1473,18 +1471,232 @@ def comp_v(self) -> int: } cs._clean() assert cs.dirty_vars == set() - assert cs.get_delta() == { - cs.get_name(): {"no_cache_v" + FIELD_MARKER: 1, "dep_v" + FIELD_MARKER: 1} - } + assert cs.get_delta() == {cs.get_name(): {"dep_v" + FIELD_MARKER: 1}} cs._clean() assert cs.dirty_vars == set() - assert cs.get_delta() == { - cs.get_name(): {"no_cache_v" + FIELD_MARKER: 1, "dep_v" + FIELD_MARKER: 1} - } + assert cs.get_delta() == {cs.get_name(): {"dep_v" + FIELD_MARKER: 1}} cs._clean() assert cs.dirty_vars == set() +def test_uncached_computed_var_unchanged_omitted_from_delta(): + """An uncached var that recomputes to the same value is left out of the delta.""" + calls = 0 + + class UncachedState(BaseState): + v: int = 0 + + @rx.var(cache=False) + def no_cache_v(self) -> int: + nonlocal calls + calls += 1 + return self.v + + ucs = UncachedState() + assert ucs.get_delta() == {ucs.get_name(): {"no_cache_v" + FIELD_MARKER: 0}} + assert calls == 1 + ucs._clean() + + # Still recomputed, but the unchanged value is not sent again. + assert ucs.get_delta() == {} + assert calls == 2 + ucs._clean() + + ucs.v = 1 + assert ucs.get_delta() == { + ucs.get_name(): {"v" + FIELD_MARKER: 1, "no_cache_v" + FIELD_MARKER: 1} + } + ucs._clean() + assert ucs.get_delta() == {} + + +def test_uncached_computed_var_scalar_key_distinguishes_types(): + """Python-equal but JSON-distinct scalars are not suppressed as unchanged.""" + values = iter([1, True, 1.0]) + + class ScalarState(BaseState): + @rx.var(cache=False) + def v(self) -> int | float: + return next(values) + + ss = ScalarState() + key = "v" + FIELD_MARKER + # 1, True and 1.0 are all Python-equal, but the client would receive 1, + # true and 1.0, so each one has to be sent. + for expected_type in (int, bool, float): + assert type(ss.get_delta()[ss.get_name()][key]) is expected_type + ss._clean() + + +def test_uncached_computed_var_nan_value_not_resent(): + """NaN is keyed by its serialized form, so an unchanged NaN is not resent.""" + + class NanState(BaseState): + @rx.var(cache=False) + def v(self) -> float: + return float("nan") + + ns = NanState() + assert math.isnan(ns.get_delta()[ns.get_name()]["v" + FIELD_MARKER]) + ns._clean() + assert ns.get_delta() == {} + + +class UncachedRedisState(BaseState): + """A state with uncached computed vars, defined at module level to be picklable.""" + + _v: int = 0 + + @rx.var(cache=False) + def scalar_v(self) -> int: + """An uncached var with an atomic value. + + Returns: + The backend var value. + """ + return self._v + + @rx.var(cache=False) + def list_v(self) -> list[int]: + """An uncached var with a value keyed by a digest. + + Returns: + A list holding the backend var value. + """ + return [self._v] + + +def test_uncached_computed_var_records_last_value_for_redis(): + """Recorded delta keys mark the state touched and survive serialization.""" + urs = UncachedRedisState() + assert urs._was_touched is False + assert urs.get_delta() == { + urs.get_name(): { + "scalar_v" + FIELD_MARKER: 0, + "list_v" + FIELD_MARKER: [0], + } + } + # The recorded keys have to reach redis, so the state counts as touched. + assert urs._was_touched is True + + # Recomputing unchanged values does not force another redis write. + urs._clean() + urs._was_touched = False + assert urs.get_delta() == {} + assert urs._was_touched is False + + # A state restored from its serialized form still knows what was sent. + restored = BaseState._deserialize(urs._serialize()) + assert isinstance(restored, UncachedRedisState) + assert restored.get_delta() == {} + + restored._v = 1 + assert restored.get_delta() == { + restored.get_name(): { + "scalar_v" + FIELD_MARKER: 1, + "list_v" + FIELD_MARKER: [1], + } + } + + +def test_uncached_computed_var_mutable_value_mutated_in_place(): + """An uncached var returning a state-owned mutable value still sees mutations.""" + + class UncachedMutableState(BaseState): + items: list[str] = [] + + @rx.var(cache=False) + def all_items(self) -> list[str]: + return self.items + + ums = UncachedMutableState() + assert ums.get_delta() == {ums.get_name(): {"all_items" + FIELD_MARKER: []}} + ums._clean() + assert ums.get_delta() == {} + ums._clean() + + ums.items.append("a") + assert ums.get_delta() == { + ums.get_name(): { + "items" + FIELD_MARKER: ["a"], + "all_items" + FIELD_MARKER: ["a"], + } + } + ums._clean() + assert ums.get_delta() == {} + + +def test_uncached_computed_var_recorded_per_client_token(): + """A value already sent to one client is still sent to another client. + + A single state instance can serve multiple clients (linked shared states), + so the recorded value only suppresses the delta for the client that got it. + """ + + class MultiClientState(BaseState): + @rx.var(cache=False) + def no_cache_v(self) -> int: + return 1 + + mcs = MultiClientState() + mcs.router = RouterData(session=SessionData(client_token="token_a")) + mcs._clean() + assert mcs.get_delta() == {mcs.get_name(): {"no_cache_v" + FIELD_MARKER: 1}} + mcs._clean() + assert mcs.get_delta() == {} + mcs._clean() + + # The same state instance now produces a delta for a different client. + mcs.router = RouterData(session=SessionData(client_token="token_b")) + mcs._clean() + assert mcs.get_delta() == {mcs.get_name(): {"no_cache_v" + FIELD_MARKER: 1}} + mcs._clean() + assert mcs.get_delta() == {} + + +def test_uncached_computed_var_unkeyable_value_always_sent(): + """A value that cannot be serialized has no key and is always sent.""" + + class CircularState(BaseState): + @rx.var(cache=False) + def circular(self) -> list: + value = [] + value.append(value) + return value + + cs = CircularState() + # Compare the keys only: the values are self-referential. + assert list(cs.get_delta()[cs.get_name()]) == ["circular" + FIELD_MARKER] + cs._clean() + assert list(cs.get_delta()[cs.get_name()]) == ["circular" + FIELD_MARKER] + + +async def test_uncached_async_computed_var_unchanged_omitted_from_delta(): + """An unchanged async uncached var is dropped from the resolved delta.""" + + class AsyncUncachedState(BaseState): + v: int = 0 + + @rx.var(cache=False) + async def no_cache_v(self) -> int: + return self.v + + aus = AsyncUncachedState() + assert await aus._get_resolved_delta() == { + aus.get_name(): {"no_cache_v" + FIELD_MARKER: 0} + } + aus._clean() + assert await aus._get_resolved_delta() == {} + aus._clean() + + aus.v = 1 + assert await aus._get_resolved_delta() == { + aus.get_name(): {"v" + FIELD_MARKER: 1, "no_cache_v" + FIELD_MARKER: 1} + } + aus._clean() + assert await aus._get_resolved_delta() == {} + + def test_computed_var_depends_on_parent_non_cached(): """Child state cached var that depends on parent state un cached var is always recalculated.""" counter = 0 @@ -3580,17 +3792,21 @@ async def test_get_state(token: str, attached_mock_event_context: EventContext): child_state2 = new_test_state.get_substate((ChildState2.get_name(),)) child_state2.value = "set_c2_value" - assert new_test_state.get_delta() == { + expected_delta = { ChildState2.get_full_name(): { "value" + FIELD_MARKER: "set_c2_value", }, GrandchildState2.get_full_name(): { "cached" + FIELD_MARKER: "set_c2_value", }, - GrandchildState3.get_full_name(): { - "computed" + FIELD_MARKER: "", - }, } + if not isinstance(state_manager, (StateManagerMemory, StateManagerDisk)): + # With redis this is a fresh instance which has not sent the uncached + # GrandchildState3.computed yet; in memory it was sent by the delta above. + expected_delta[GrandchildState3.get_full_name()] = { + "computed" + FIELD_MARKER: "", + } + assert new_test_state.get_delta() == expected_delta @pytest.mark.asyncio