From fc2470c260434f28c3546c47f20fd1d1ba02e4ce Mon Sep 17 00:00:00 2001 From: Farhan Ali Raza <62690310+FarhanAliRaza@users.noreply.github.com> Date: Wed, 5 Aug 2026 19:14:32 +0500 Subject: [PATCH 1/3] feat(reflex): chart plans and @reflex_xy.data column vars MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two halves the data-bound component tier is made of, with no wire or component surface yet. plan.py — a ChartPlan is a validated, data-free chart structure. Building one constructs the real xy tree from string channels, binds zero-row placeholder columns for every referenced channel through the production resolution path (a recording table, so the column list cannot drift from what binding will look up), and runs .figure() once: the full mark/config validation gate, in milliseconds, with no data ingestion. The canonical JSON (plan_version: 1) is content-addressed into a digest and registered in a process-local map; binding is columns + plan -> a fresh Chart -> figure, with mismatch errors that name both sides. PLAN_VERSION and a golden digest are pinned so accidental format churn fails a test rather than silently invalidating every mounted client's digest. data_vars.py — @reflex_xy.data is the exact sibling of @reflex_xy.figure: same builder-targeted dependency tracking, pre-session short-circuit, underscore refusal, async dispatch, and None-releases semantics, but its value is a DataHandle and what it publishes is columns. The method's return annotation is the schema channel: a TypedDict parametrizes the handle, so the layer above can read column names without executing user code. tokens.py grows two families beside xyv1: xyd1 for data tokens (same grammar and rebuild contract) and the composite xyp1|| that names a bound figure — plan digest for the structure, data token for the columns, both halves independently recoverable. registry.py stores column entries as their own kind: immutable generations with no locks and no pins (a republish replaces the whole entry), swept under the existing TTL because the data method is the rebuild recipe. The data-token -> {digest} index and the on_error seam land here; the namespace wires them in the next change. Spec: reflex-integration.md §3.6 (data vars, plans, column entries, format stability), file map. --- python/reflex_xy/__init__.py | 7 + python/reflex_xy/data_vars.py | 233 +++++++++++++++++++++++++ python/reflex_xy/plan.py | 237 ++++++++++++++++++++++++++ python/reflex_xy/registry.py | 159 ++++++++++++++++- python/reflex_xy/tokens.py | 117 ++++++++++--- spec/design/reflex-integration.md | 90 +++++++++- tests/reflex_adapter/test_data_var.py | 219 ++++++++++++++++++++++++ tests/reflex_adapter/test_plan.py | 113 ++++++++++++ 8 files changed, 1142 insertions(+), 33 deletions(-) create mode 100644 python/reflex_xy/data_vars.py create mode 100644 python/reflex_xy/plan.py create mode 100644 tests/reflex_adapter/test_data_var.py create mode 100644 tests/reflex_adapter/test_plan.py diff --git a/python/reflex_xy/__init__.py b/python/reflex_xy/__init__.py index 0a554722..7e184303 100644 --- a/python/reflex_xy/__init__.py +++ b/python/reflex_xy/__init__.py @@ -54,6 +54,9 @@ def index() -> rx.Component: "set_view": ".app", "setup": ".app", "chart": ".component", + "AsyncDataVar": ".data_vars", + "DataVar": ".data_vars", + "data": ".data_vars", "DataHandle": ".handles", "FigureHandle": ".handles", "CanonicalRowIdGroup": ".events", @@ -78,10 +81,12 @@ def index() -> rx.Component: __all__ = [ "XY_NAMESPACE", + "AsyncDataVar", "AsyncFigureVar", "CanonicalRowIdGroup", "DataBounds", "DataHandle", + "DataVar", "FigureHandle", "FigureRegistry", "FigureVar", @@ -98,6 +103,7 @@ def index() -> rx.Component: "append", "chart", "clear_selection", + "data", "figure", "inline", "register", @@ -231,6 +237,7 @@ def release(token: "str | FigureHandle") -> None: if TYPE_CHECKING: from .app import XYPlugin, append, clear_selection, reset_view, select, set_view, setup from .component import chart + from .data_vars import AsyncDataVar, DataVar, data from .events import ( CanonicalRowIdGroup, DataBounds, diff --git a/python/reflex_xy/data_vars.py b/python/reflex_xy/data_vars.py new file mode 100644 index 00000000..2dd72fa7 --- /dev/null +++ b/python/reflex_xy/data_vars.py @@ -0,0 +1,233 @@ +"""`@reflex_xy.data`: a computed var that *is* the dataset registration. + +The data-plane sibling of `@reflex_xy.figure` (vars.py) for the data-bound +component tier: the state method returns **columns only** — a mapping of +column name to array-likes — so there is no chart API inside it to get +wrong. Evaluating the var publishes the columns into the per-process +registry under a deterministic token (`xyd1|||`) and +its value is a tiny typed :class:`~reflex_xy.handles.DataHandle`. Reflex's +dependency tracking watches the method body; a state change republishes the +columns, and the registry rebuilds + broadcasts every mounted chart plan +bound to them. + +The method's return annotation is the compile-time schema channel (fact +R7): annotate a ``TypedDict`` and the class-level var carries +``DataHandle[Schema]``, which the chart factories read column names from — +without executing any user code. A plain ``dict[str, ...]`` annotation +degrades gracefully to first-execution validation. + +Like figure builders, data methods must be pure functions of their state +instance: the token is the rebuild recipe (state_bridge.py re-runs the +method when a fresh worker needs the columns back), and purity is what +makes the column set a rebuildable cache instead of precious process state. +""" + +from __future__ import annotations + +import inspect +from collections.abc import Callable, Mapping +from typing import Any, Optional, get_type_hints, is_typeddict, overload + +from reflex_base.vars.base import AsyncComputedVar, ComputedVar + +from .handles import DataHandle +from .registry import registry +from .tokens import BUILDER_ATTR, build_data_token + +__all__ = ["AsyncDataVar", "DataVar", "data", "validate_columns"] + + +def _builder_target(var: Any, obj: Any) -> Any: + """Point dependency tracking at the *data method*, not the token wrapper + (same reason as vars.py: the wrapper fget reads nothing but the router).""" + if obj is not None: + return obj + return getattr(var._fget, BUILDER_ATTR, None) + + +class DataVar(ComputedVar): + """ComputedVar whose value is a DataHandle (sync data method).""" + + def _deps(self, objclass: Any, obj: Any = None) -> dict[str, set[str]]: + return ComputedVar._deps(self, objclass, obj=_builder_target(self, obj)) + + +class AsyncDataVar(AsyncComputedVar): + """AsyncComputedVar whose value is a DataHandle (async data method).""" + + def _deps(self, objclass: Any, obj: Any = None) -> dict[str, set[str]]: + return AsyncComputedVar._deps(self, objclass, obj=_builder_target(self, obj)) + + +def validate_columns(columns: Any, *, source: str) -> dict[str, Any]: + """The only checks that need real data: a mapping of named, equal-length + array-like columns. Everything structural was validated at compile by the + plan's zero-row probe; dtype/shape details stay with figure compilation.""" + if not isinstance(columns, Mapping): + raise TypeError( + f"{source} must return a mapping of column name -> values " + f"(e.g. a TypedDict of arrays), got {type(columns).__name__}" + ) + validated: dict[str, Any] = {} + lengths: dict[str, int] = {} + for key, values in columns.items(): + if not isinstance(key, str): + raise TypeError(f"{source} column names must be strings, got {key!r}") + if isinstance(values, (str, bytes, Mapping)): + raise TypeError( + f"{source} column {key!r} must be an array-like of values, " + f"got {type(values).__name__}" + ) + try: + lengths[key] = len(values) + except TypeError as exc: + raise TypeError( + f"{source} column {key!r} must be an array-like with a length, " + f"got {type(values).__name__}" + ) from exc + validated[key] = values + if len(set(lengths.values())) > 1: + detail = ", ".join(f"{key}={length}" for key, length in lengths.items()) + raise ValueError(f"{source} columns must share one length, got {detail}") + return validated + + +def _mint_token(state: Any, var_name: str) -> Optional[str]: + client_token = state.router.session.client_token + if not client_token: + return None + return build_data_token(client_token, type(state).get_full_name(), var_name) + + +def _publish(token: str, columns: Any, *, source: str) -> DataHandle: + if columns is None: + registry.release_columns(token) + return DataHandle("") + registry.publish_columns(token, validate_columns(columns, source=source)) + return DataHandle(token) + + +def _source_label(method: Callable[..., Any]) -> str: + qualname = getattr(method, "__qualname__", None) or getattr(method, "__name__", "data method") + return qualname.rsplit("..", 1)[-1] + + +def _adopt_identity(fget: Any, method: Callable[..., Any], name: str) -> None: + fget.__name__ = name + fget.__qualname__ = getattr(method, "__qualname__", name) + fget.__module__ = getattr(method, "__module__", fget.__module__) + fget.__doc__ = method.__doc__ + setattr(fget, BUILDER_ATTR, method) + + +def _make_fget(method: Callable[[Any], Any]) -> Callable[[Any], DataHandle]: + name = _fn_name(method) + source = _source_label(method) + + def fget(self: Any) -> DataHandle: + token = _mint_token(self, name) + if token is None: + return DataHandle("") + return _publish(token, method(self), source=source) + + _adopt_identity(fget, method, name) + return fget + + +def _make_async_fget(method: Callable[[Any], Any]) -> Callable[[Any], Any]: + name = _fn_name(method) + source = _source_label(method) + + async def fget(self: Any) -> DataHandle: + token = _mint_token(self, name) + if token is None: + return DataHandle("") + return _publish(token, await method(self), source=source) + + _adopt_identity(fget, method, name) + return fget + + +def _fn_name(fn: Callable[..., Any]) -> str: + name = getattr(fn, "__name__", "") + if not name: + msg = f"@reflex_xy.data methods must be named functions, got {fn!r}" + raise TypeError(msg) + return name + + +def _return_type(method: Callable[..., Any]) -> Any: + """``DataHandle[Schema]`` when the return annotation is a TypedDict — + the schema survives as the class-level var's ``_var_type`` (R7) and is + how the factories compile-check column names. Anything else (plain + dicts, missing or unresolvable annotations) degrades to ``DataHandle``: + columns are then validated on first execution instead.""" + try: + hints = get_type_hints(method) + except Exception: # noqa: BLE001 - annotations may reference unimportable names + return DataHandle + annotation = hints.get("return") + if annotation is not None and is_typeddict(annotation): + return DataHandle[annotation] + return DataHandle + + +@overload +def data(method: Callable[[Any], Any]) -> "DataVar | AsyncDataVar": ... + + +@overload +def data( + method: None = None, **var_kwargs: Any +) -> Callable[[Callable[[Any], Any]], "DataVar | AsyncDataVar"]: ... + + +def data( + method: Optional[Callable[[Any], Any]] = None, **var_kwargs: Any +) -> "DataVar | AsyncDataVar | Callable[[Callable[[Any], Any]], DataVar | AsyncDataVar]": + """Declare a chart dataset on a Reflex state class. + + Usage:: + + class CloudData(TypedDict): + x: np.ndarray + y: np.ndarray + mag: np.ndarray + + class Dash(rx.State): + points: int = 200_000 + + @reflex_xy.data + def cloud(self) -> CloudData: + rng = np.random.default_rng(7) + x = rng.normal(size=self.points) + return {"x": x, "y": x * 0.6, "mag": np.abs(x)} + + # in the page: + # reflex_xy.scatter_chart(data=Dash.cloud, x="x", y="y", color="mag") + + The method must return a mapping of column name -> equal-length + array-likes, or ``None`` for "no data right now" (which releases the + registered columns and yields the empty handle). ``async def`` methods + become ``AsyncDataVar``s (same dispatch rule as ``rx.var``); keyword + arguments pass through to reflex's computed var (``deps=``, + ``interval=``, ...). + """ + + def _decorate(fn: Callable[[Any], Any]) -> "DataVar | AsyncDataVar": + if _fn_name(fn).startswith("_"): + # Same rule as figure vars: the handle must sync to the client, + # and backend (underscore) vars never do. + msg = ( + "@reflex_xy.data vars must not start with '_' (the handle must sync to the client)" + ) + raise ValueError(msg) + var_kwargs.setdefault("cache", True) + return_type = _return_type(fn) + if inspect.iscoroutinefunction(fn): + return AsyncDataVar(fget=_make_async_fget(fn), return_type=return_type, **var_kwargs) + return DataVar(fget=_make_fget(fn), return_type=return_type, **var_kwargs) + + if method is None: + return _decorate + return _decorate(method) diff --git a/python/reflex_xy/plan.py b/python/reflex_xy/plan.py new file mode 100644 index 00000000..4e828924 --- /dev/null +++ b/python/reflex_xy/plan.py @@ -0,0 +1,237 @@ +"""Chart plans: validated, data-free chart structure for the data-bound tier. + +A plan is the server-side half of the composite figure identity +``xyp1||`` (spec/design/reflex-integration.md): the xy +node tree with **string channels only**, compiled once at page evaluation. + +- **Build** (factory call = page evaluation = Reflex compile): construct the + real xy tree, bind a zero-row placeholder column for every referenced + channel name, and call ``.figure()`` once — the full mark/config + validation gate (facts X1/X2, pinned in tests/test_validation_timing.py) + runs in milliseconds with no real data. The probe figure is discarded; + what is kept is the digest, the recorded column names, and (for live + charts) the probe's Tailwind class inventory. +- **Serialize**: nodes → canonical JSON (sorted keys, ``plan_version``) → + sha256 prefix = ``digest``. The digest is a *content address*: every + worker that evaluates the page derives the same digest and holds the plan + in this process-local map — and backend-only workers, which skip the + frontend compile, get the same evaluation from ``setup(app)``'s startup + lifespan (app.py `_ensure_page_plans`). A lookup miss is hot-reload + drift and answers ``err {resync}`` naming the digest. +- **Bind** (serve time): columns + plan → a **fresh** ``Chart`` (never + reuse — ``Chart.figure()`` memoizes, X3) → ``Figure``. Column mismatches + raise :class:`PlanBindError` naming both sides. + +Plans hold no data and no session identity; they are pure functions of page +source. Everything session-shaped lives in the data token half. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +from collections.abc import Mapping +from typing import Any, Optional + +import numpy as np + +from xy.components import Chart, Component, Mark + +__all__ = [ + "PLAN_VERSION", + "ChartPlan", + "PlanBindError", + "PlanError", + "PlanMissError", + "build_plan", + "plan_of", + "register_plan", +] + +PLAN_VERSION = 1 +_DIGEST_CHARS = 20 # sha256 hex prefix; content address for a process-local map + +#: Mark kinds whose figure-compile validators require at least one finite +#: value (they aggregate: quantiles, bins, meshes). The zero-row probe +#: cannot compile them, so they are excluded from the plan tier — the +#: Phase 3 decision recorded in reflex-component-api-implementation.md. +_NEEDS_DATA_MARKS = frozenset({"box", "violin", "hexbin", "contour", "heatmap", "stairs", "ecdf"}) + + +class PlanError(ValueError): + """A chart plan could not be built, resolved, or bound.""" + + +class PlanMissError(PlanError): + """No plan registered under a digest (hot-reload drift — client resyncs).""" + + def __init__(self, digest: str) -> None: + super().__init__( + f"unknown chart plan {digest!r} on this worker (stale page after a " + "hot reload?); re-subscribe against the recompiled page" + ) + self.digest = digest + + +class PlanBindError(PlanError): + """The data var's columns do not satisfy the plan's bindings.""" + + +class _ProbeTable(Mapping): + """Zero-row placeholder table that records every column it resolves. + + ``Chart.figure()`` resolves string channels through ``data[name]`` + (the exact production code path), so the recorded names are *derived* + from the real resolution logic — the plan's column list can never drift + from what binding will actually look up. + """ + + def __init__(self) -> None: + self.seen: list[str] = [] + + def __getitem__(self, key: str) -> np.ndarray: + if key not in self.seen: + self.seen.append(key) + return np.empty(0, dtype=np.float64) + + def __iter__(self): # pragma: no cover - Mapping protocol completeness + return iter(self.seen) + + def __len__(self) -> int: # pragma: no cover - Mapping protocol completeness + return len(self.seen) + + +def _plain(value: Any, context: str) -> Any: + """Canonical JSON-able copy of one plan node field (fail closed).""" + if dataclasses.is_dataclass(value) and not isinstance(value, type): + node: dict[str, Any] = {"~node": type(value).__name__} + for field in dataclasses.fields(value): + node[field.name] = _plain(getattr(value, field.name), f"{context}.{field.name}") + return node + if isinstance(value, Mapping): + return {str(key): _plain(item, f"{context}.{key}") for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_plain(item, context) for item in value] + if isinstance(value, np.generic): + return value.item() + if isinstance(value, (str, int, float, bool)) or value is None: + return value + raise PlanError( + f"{context} holds a {type(value).__name__}, which cannot be part of a " + "data-bound chart plan. Plans are data-free structure: bind columns " + "by name (strings) and keep arrays in the @reflex_xy.data method; " + "components (e.g. legend/tooltip render=) belong to the escape " + "hatch (@reflex_xy.figure) or the static tier." + ) + + +@dataclasses.dataclass(frozen=True) +class ChartPlan: + """One validated, data-free chart structure, addressed by content.""" + + kind: str + children: tuple[Component, ...] + chart_props: dict[str, Any] + columns: tuple[str, ...] # channel names the probe resolved, in order + tailwind_classes: str # probe figure's DOM class inventory (live-tier scan) + digest: str + + def chart(self, data: Any) -> Chart: + """A fresh ``Chart`` over ``data`` (X3: never reuse a compiled one).""" + return Chart(self.kind, self.children, data=data, **self.chart_props) + + def bind(self, columns: Mapping[str, Any], *, source: str = "the data var") -> Chart: + """Bind real columns; missing bindings name both sides.""" + missing = [name for name in self.columns if name not in columns] + if missing: + bound = ", ".join(repr(name) for name in missing) + produced = ", ".join(sorted(str(key) for key in columns)) or "no columns" + noun = "columns" if len(missing) > 1 else "column" + raise PlanBindError(f"plan binds {noun} {bound}; {source} produced {{{produced}}}") + return self.chart(columns) + + +def build_plan( + kind: str, children: tuple[Component, ...], chart_props: dict[str, Any] +) -> ChartPlan: + """Compile + validate a plan and register it in this worker's map.""" + for child in children: + if isinstance(child, Mark) and child.data is not None: + raise PlanError( + "per-mark data= is not supported in data-bound charts; bind one " + "chart-level data source (this is tracked as deferred work in " + "spec/design/reflex-component-api-implementation.md)" + ) + serialized = { + "plan_version": PLAN_VERSION, + "kind": kind, + "chart": {key: _plain(value, f"{kind}() {key}") for key, value in chart_props.items()}, + "children": [_plain(child, f"{kind}() child {i}") for i, child in enumerate(children)], + } + canonical = json.dumps(serialized, sort_keys=True, separators=(",", ":")) + digest = hashlib.sha256(canonical.encode()).hexdigest()[:_DIGEST_CHARS] + + # The compile-time validation gate: bind zero-row placeholders for every + # string channel and compile once. Errors surface here — at page + # evaluation — with the ordinary xy messages. + probe = _ProbeTable() + try: + probe_figure = Chart(kind, children, data=probe, **chart_props).figure() + except ValueError as exc: + needy = sorted( + { + child.kind + for child in children + if isinstance(child, Mark) and child.kind in _NEEDS_DATA_MARKS + } + ) + if needy: + raise PlanError( + f"{', '.join(needy)} marks aggregate their values, so their " + "validators need at least one row — the zero-row plan probe " + "cannot compile them. Data-bound charts exclude these kinds " + "(recorded in reflex-component-api-implementation.md, Phase 3 " + "decision); build the chart with @reflex_xy.figure, or pass " + "a concrete xy Chart to reflex_xy.chart() for the static tier." + ) from exc + raise + tailwind_classes = " ".join(probe_figure.dom_class_strings()) + + plan = ChartPlan( + kind=kind, + children=children, + chart_props=chart_props, + columns=tuple(probe.seen), + tailwind_classes=tailwind_classes, + digest=digest, + ) + return register_plan(plan) + + +#: Process-local {digest: plan}, populated wherever page bodies evaluate: +#: the frontend compile, and — for backend-only workers that skip it — the +#: setup(app) startup lifespan (app.py `_ensure_page_plans`). Entries are +#: tiny (node dataclasses with string channels) and bounded by page code. +_PLANS: dict[str, ChartPlan] = {} + + +def register_plan(plan: ChartPlan) -> ChartPlan: + """Idempotently register a plan under its digest; returns the canonical one.""" + return _PLANS.setdefault(plan.digest, plan) + + +def plan_of(digest: str) -> Optional[ChartPlan]: + return _PLANS.get(digest) + + +def require_plan(digest: str) -> ChartPlan: + plan = _PLANS.get(digest) + if plan is None: + raise PlanMissError(digest) + return plan + + +def reset_plans_for_tests() -> None: + """Forget every registered plan (test isolation only).""" + _PLANS.clear() diff --git a/python/reflex_xy/registry.py b/python/reflex_xy/registry.py index 8986923d..3c394246 100644 --- a/python/reflex_xy/registry.py +++ b/python/reflex_xy/registry.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import logging import threading import time import uuid @@ -28,7 +29,9 @@ if TYPE_CHECKING: from xy._figure import Figure -__all__ = ["FigureEntry", "FigureRegistry", "registry"] +__all__ = ["ColumnEntry", "FigureEntry", "FigureRegistry", "registry"] + +_logger = logging.getLogger("reflex_xy") # Idle figures are swept after this long without a subscribe/message/publish. # Deterministic (state-backed) figures rebuild transparently on the next @@ -94,11 +97,38 @@ def touch(self) -> None: self.last_access = time.monotonic() +@dataclass +class ColumnEntry: + """One registered column set (the data half of the data-bound tier). + + Column entries are pure rebuildable caches of Reflex state — the + `@reflex_xy.data` method is the recipe — so they carry none of the + figure entry's kernel machinery: no locks (a republish replaces the + whole immutable generation), no pins, always sweepable. + """ + + columns: dict[str, Any] + token: str + version: int = 1 + last_access: float = field(default_factory=time.monotonic) + + def touch(self) -> None: + self.last_access = time.monotonic() + + class FigureRegistry: """token -> FigureEntry map with versioning and TTL sweep.""" def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None: self._entries: dict[str, FigureEntry] = {} + # Data-bound tier: column sets published by @reflex_xy.data vars, + # and the data-token -> {plan digest} index that lets a column + # republish rebuild + broadcast every mounted dependent figure. The + # index is bounded by mounted plans: entries are added when a + # composite figure binds and dropped when a republish finds neither + # a cached figure nor live subscribers under the composite token. + self._column_entries: dict[str, ColumnEntry] = {} + self._digests_by_data_token: dict[str, set[str]] = {} # A mounted client remains in its socket room when a TTL sweep evicts # a rebuildable figure. Retain only the evicted scalar version while # at least one such client is subscribed, so a state-driven rebuild on @@ -134,6 +164,11 @@ def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None: # async callback(token, message, buffers, version) -> None for append # and view-state pushes — the message-shaped data-plane seam. self._on_push: Optional[PushHook] = None + # async callback(token, error, resync) -> None: room-wide err frames + # for server-side failures with no request to answer (a data + # republish whose plan bind fails must not leave stale pixels + # silently frozen). + self._on_error: Optional[Callable[[str, str, bool], Awaitable[None]]] = None # -- wiring ------------------------------------------------------------ @@ -149,6 +184,28 @@ def on_push( ) -> None: self._on_push = callback + def on_error(self, callback: Callable[[str, str, bool], Awaitable[None]]) -> None: + self._on_error = callback + + def _schedule_error(self, token: str, error: str, *, resync: bool) -> None: + """Fan an out-of-band failure to a figure room from any thread.""" + callback = self._on_error + loop = self._loop + if callback is None or loop is None: + return + + async def _run() -> None: + await callback(token, error, resync) + + try: + running = asyncio.get_running_loop() + except RuntimeError: + running = None + if running is loop: + loop.create_task(_run()) + else: + asyncio.run_coroutine_threadsafe(_run(), loop) + def _enqueue_push( self, entry: FigureEntry, @@ -478,6 +535,92 @@ def __len__(self) -> int: with self._mutex: return len(self._entries) + # -- data-bound tier: column entries + plan index ------------------------ + + def publish_columns(self, token: str, columns: dict[str, Any]) -> ColumnEntry: + """Insert or replace a column set, then rebuild every mounted plan + bound to it (fresh figures under the composite tokens, broadcast by + the ordinary publish fan-out). Entries are immutable generations: a + republish replaces the whole entry and bumps its version.""" + with self._mutex: + previous = self._column_entries.get(token) + version = 1 if previous is None else previous.version + 1 + entry = ColumnEntry(columns=columns, token=token, version=version) + self._column_entries[token] = entry + digests = sorted(self._digests_by_data_token.get(token, ())) + for digest in digests: + self._rebuild_dependent(token, digest, columns) + return entry + + def get_columns(self, token: str) -> Optional[ColumnEntry]: + with self._mutex: + entry = self._column_entries.get(token) + if entry is not None: + entry.touch() + return entry + + def release_columns(self, token: str) -> None: + """Drop a column set and every dependent composite figure entry. + + The "no data right now" path (`@reflex_xy.data` returning None): the + empty handle unmounts subscribed charts client-side, and dropping + the dependent figure caches here keeps a later re-mount rebuilding + from current state instead of serving pre-release columns. + """ + from .tokens import build_plan_token + + with self._mutex: + self._column_entries.pop(token, None) + digests = self._digests_by_data_token.pop(token, set()) + for digest in sorted(digests): + self.release(build_plan_token(digest, token)) + + def bind_plan(self, data_token: str, digest: str) -> None: + """Record that a mounted plan binds this data token (idempotent).""" + with self._mutex: + self._digests_by_data_token.setdefault(data_token, set()).add(digest) + + def _rebuild_dependent(self, data_token: str, digest: str, columns: dict[str, Any]) -> None: + """Re-bind one dependent plan against freshly published columns.""" + from .plan import plan_of + from .tokens import build_plan_token, parse_token + + composite = build_plan_token(digest, data_token) + with self._mutex: + mounted = composite in self._entries or bool( + self._rebuildable_subscribers.get(composite) + ) + if not mounted: + # Nothing serves or watches this plan anymore: forget the + # binding (this is what keeps the index bounded by mounts); + # a later subscribe rebuilds from scratch and re-indexes. + digests = self._digests_by_data_token.get(data_token) + if digests is not None: + digests.discard(digest) + if not digests: + self._digests_by_data_token.pop(data_token, None) + return + parsed = parse_token(data_token) + source = ( + f"{parsed.state_full_name}.{parsed.var_name}" if parsed is not None else "the data var" + ) + plan = plan_of(digest) + try: + if plan is None: + from .plan import PlanMissError + + raise PlanMissError(digest) + figure = plan.bind(columns, source=source).figure() + except Exception as exc: # noqa: BLE001 - user data meets page structure here + # Fail loud on both sides: the server log carries the bind error, + # and subscribers get an err frame instead of silently frozen + # pixels (their bounded resync retries against current state). + _logger.warning("republish of %s failed: %s", composite, exc) + self.release(composite) + self._schedule_error(composite, str(exc), resync=True) + return + self.publish(composite, figure) + # -- version bump + fan-out --------------------------------------------- def bump(self, token: str, *, expected: Optional[FigureEntry] = None) -> Optional[FigureEntry]: @@ -703,7 +846,12 @@ def clear_selection(self, token: str) -> None: # -- TTL sweep ----------------------------------------------------------- def sweep(self, *, now: Optional[float] = None) -> list[str]: - """Drop unpinned entries idle past the TTL; returns dropped tokens.""" + """Drop unpinned entries idle past the TTL; returns dropped tokens. + + Column entries sweep under the same TTL: they are state-rebuildable + caches exactly like state-token figures (a later subscribe re-runs + the data method), so the sweep bounds column memory, not correctness. + """ now = time.monotonic() if now is None else now dropped: list[str] = [] with self._mutex: @@ -717,6 +865,10 @@ def sweep(self, *, now: Optional[float] = None) -> list[str]: self._retain_removed_version_locked(token, entry) del self._entries[token] dropped.append(token) + for token, column_entry in list(self._column_entries.items()): + if now - column_entry.last_access > self._ttl: + del self._column_entries[token] + dropped.append(token) return dropped async def sweep_forever(self) -> None: @@ -734,6 +886,8 @@ async def sweep_forever(self) -> None: def reset_registry_for_tests() -> FigureRegistry: """Reset the process registry in place (test isolation only).""" registry._entries.clear() + registry._column_entries.clear() + registry._digests_by_data_token.clear() registry._evicted_versions.clear() registry._rebuildable_subscribers.clear() registry._rebuildable_tokens_by_sid.clear() @@ -742,6 +896,7 @@ def reset_registry_for_tests() -> FigureRegistry: registry._loop = None registry._on_publish = None registry._on_push = None + registry._on_error = None registry._ttl = DEFAULT_TTL_SECONDS return registry diff --git a/python/reflex_xy/tokens.py b/python/reflex_xy/tokens.py index 049dad5f..9aed0308 100644 --- a/python/reflex_xy/tokens.py +++ b/python/reflex_xy/tokens.py @@ -1,22 +1,31 @@ -"""Figure tokens: the only chart-related value that lives in Reflex state. - -Two token families, one namespace: - -- **State tokens** (`xyv1|||`) are - minted by the `@reflex_xy.figure` computed var. They are *deterministic*: - any backend worker holding the same Reflex state can re-derive the figure - from the token alone, which is what makes reconnects and multi-worker - deployments work without a central figure store (the token IS the recipe; - Reflex state is the pantry). -- **Opaque tokens** (`xyfig-`) come from imperative - `reflex_xy.register(...)`. They cannot be rebuilt elsewhere — dev-tier by - design, documented in spec/design/reflex-integration.md. +"""Chart tokens: the only chart-related values that live in Reflex state. + +Three token families, one namespace: + +- **Figure state tokens** (`xyv1|||`) + are minted by the `@reflex_xy.figure` computed var. They are + *deterministic*: any backend worker holding the same Reflex state can + re-derive the figure from the token alone, which is what makes reconnects + and multi-worker deployments work without a central figure store (the + token IS the recipe; Reflex state is the pantry). +- **Data state tokens** (`xyd1|||`) + are minted by the `@reflex_xy.data` computed var under the same grammar + and rebuild contract, naming a registered *column set* instead of a + figure. They never subscribe alone: the wrapper composes them with a plan + digest into a **plan token** (`xyp1||`), the composite + figure identity of the data-bound tier — plan digest for the structure, + data token for the columns; both halves independently recoverable. +- **Opaque tokens** (`xyfig-` / `xyin-`) come from imperative + `reflex_xy.register(...)` / `inline(...)`. They cannot be rebuilt + elsewhere — dev/fixed tiers by design, documented in + spec/design/reflex-integration.md. Tokens are visible to their own client (they ride through state deltas), so they must not carry anything the client doesn't already know: the client -token is the browser tab's own session id, and state/var names already -appear in every state delta. Cross-client use is refused by the namespace's -affinity check, not by token secrecy. +token is the browser tab's own session id, state/var names already appear +in every state delta, and a plan digest is a content address of page code +the client was compiled from. Cross-client use is refused by the +namespace's affinity check, not by token secrecy. """ from __future__ import annotations @@ -27,23 +36,32 @@ from typing import Any, Optional __all__ = [ + "ParsedPlanToken", "ParsedToken", + "build_data_token", + "build_plan_token", "build_state_token", "builder_of", + "parse_plan_token", "parse_token", ] -_PREFIX = "xyv1" +_FIGURE_PREFIX = "xyv1" +_DATA_PREFIX = "xyd1" +_PLAN_PREFIX = "xyp1" _SEP = "|" # Client tokens are UUID-ish; state full names and var names are dotted # Python identifiers. Nothing here may contain the separator. -_TOKEN_RE = re.compile( - r"^xyv1\|(?P[A-Za-z0-9_-]{8,64})" +_STATE_TOKEN_BODY = ( + r"(?P[A-Za-z0-9_-]{8,64})" r"\|(?P[A-Za-z0-9_.]{1,512})" r"\|(?P[A-Za-z_][A-Za-z0-9_]{0,255})$" ) +_TOKEN_RE = re.compile(r"^(?Pxyv1|xyd1)\|" + _STATE_TOKEN_BODY) +# Plan digests are lowercase sha256 hex prefixes (plan.py). +_PLAN_RE = re.compile(r"^xyp1\|(?P[0-9a-f]{8,64})\|(?Pxyd1\|.+)$") -#: Attribute stashed on a figure var's fget carrying the user's builder. +#: Attribute stashed on a figure/data var's fget carrying the user's builder. #: It lives on the *function* (not the ComputedVar) so it survives reflex's #: `_replace` copies, which re-instantiate the var but thread fget through. BUILDER_ATTR = "__xy_builder__" @@ -51,23 +69,55 @@ @dataclass(frozen=True) class ParsedToken: + """A parsed state token; ``kind`` is ``"figure"`` (xyv1) or ``"data"`` (xyd1).""" + client_token: str state_full_name: str var_name: str + kind: str = "figure" -def build_state_token(client_token: str, state_full_name: str, var_name: str) -> str: - token = _SEP.join((_PREFIX, client_token, state_full_name, var_name)) - if parse_token(token) is None: +@dataclass(frozen=True) +class ParsedPlanToken: + """A parsed composite plan token: plan digest + the embedded data token.""" + + digest: str + data: ParsedToken + data_token: str # the verbatim xyd1|… substring (registry key for columns) + + +def _build_state_token(prefix: str, client_token: str, state_full_name: str, var_name: str) -> str: + token = _SEP.join((prefix, client_token, state_full_name, var_name)) + if _TOKEN_RE.match(token) is None: # Defensive: a state or client token that defeats the grammar would # otherwise mint a token the namespace can never resolve. - msg = f"cannot build a valid figure token from {client_token!r}/{state_full_name!r}/{var_name!r}" + msg = f"cannot build a valid {prefix} token from {client_token!r}/{state_full_name!r}/{var_name!r}" + raise ValueError(msg) + return token + + +def build_state_token(client_token: str, state_full_name: str, var_name: str) -> str: + """Mint a figure token (`xyv1|…`).""" + return _build_state_token(_FIGURE_PREFIX, client_token, state_full_name, var_name) + + +def build_data_token(client_token: str, state_full_name: str, var_name: str) -> str: + """Mint a data token (`xyd1|…`).""" + return _build_state_token(_DATA_PREFIX, client_token, state_full_name, var_name) + + +def build_plan_token(digest: str, data_token: str) -> str: + """Compose the data-bound tier's figure identity (`xyp1||`).""" + token = _SEP.join((_PLAN_PREFIX, digest, data_token)) + if parse_plan_token(token) is None: + msg = f"cannot build a valid plan token from {digest!r}/{data_token!r}" raise ValueError(msg) return token def parse_token(token: str) -> Optional[ParsedToken]: - """Parse a state token; None for opaque/foreign strings (fail closed).""" + """Parse a state token (figure or data); None for opaque/foreign strings + (fail closed). Composite plan tokens parse via :func:`parse_plan_token`.""" if not isinstance(token, str): return None match = _TOKEN_RE.match(token) @@ -77,11 +127,26 @@ def parse_token(token: str) -> Optional[ParsedToken]: client_token=match["client"], state_full_name=match["state"], var_name=match["var"], + kind="figure" if match["prefix"] == _FIGURE_PREFIX else "data", ) +def parse_plan_token(token: str) -> Optional[ParsedPlanToken]: + """Parse a composite plan token; None for anything else (fail closed).""" + if not isinstance(token, str): + return None + match = _PLAN_RE.match(token) + if match is None: + return None + data = parse_token(match["data"]) + if data is None or data.kind != "data": + return None + return ParsedPlanToken(digest=match["digest"], data=data, data_token=match["data"]) + + def builder_of(state_cls: Any, var_name: str) -> Optional[Callable[[Any], Any]]: - """Find the figure builder a `@reflex_xy.figure` var attached to a state class.""" + """Find the builder a `@reflex_xy.figure` / `@reflex_xy.data` var + attached to a state class.""" computed = getattr(state_cls, "computed_vars", None) var = computed.get(var_name) if isinstance(computed, dict) else None fget = getattr(var, "_fget", None) diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md index 23b4a9ec..fe9927f1 100644 --- a/spec/design/reflex-integration.md +++ b/spec/design/reflex-integration.md @@ -425,6 +425,82 @@ generations are leased; the sweep skips them until the mutation and version bump finish. Rapid re-publishes coalesce: an un-started broadcast absorbs newer publishes and always ships the latest payload. +### 3.6 The data-bound tier: `@reflex_xy.data`, plans, composite tokens + +The primary component API (adopted design: Option 6 of +[`reflex-component-api-options.md`](reflex-component-api-options.md); work +plan in [`reflex-component-api-implementation.md`](reflex-component-api-implementation.md)) +splits the figure var's job in two: **structure is declared in the page**, +**state supplies only columns**. + +```python +class CloudData(TypedDict): + x: np.ndarray; y: np.ndarray; mag: np.ndarray + +class Dash(rx.State): + points: int = 200_000 + + @reflex_xy.data # columns only — no chart API inside + def cloud(self) -> CloudData: ... + +def index(): + return reflex_xy.scatter_chart( # flat form; reflex_xy.chart(*nodes) + data=Dash.cloud, # is the composed multi-mark form + x="x", y="y", color="mag", colormap="viridis", + height="460px", on_select_end=Dash.select, + ) +``` + +**`@reflex_xy.data`** (`data_vars.py` — the module is named `data_vars` +because a `data.py` submodule would shadow the `reflex_xy.data` export) is +the exact sibling of `@reflex_xy.figure`: a computed var whose value is a +typed `DataHandle` wrapping `xyd1|||` (same grammar, +charset, purity contract, pre-session short-circuit, underscore refusal, +async dispatch, and `None`-releases semantics as figure vars). Evaluating +it validates the returned mapping — string keys, array-likes, one shared +length; the only checks that need real data — and publishes the **columns** +into the registry. The method's return annotation is the compile-time +schema channel (fact R7): a `TypedDict` parametrizes the handle +(`DataHandle[CloudData]`), and the factories read the column names from the +class-level var without executing anything; a plain `dict` annotation +degrades to first-execution validation. + +**Plans** (`plan.py`). A chart factory call at page evaluation compiles its +xy nodes (string channels only) into a `ChartPlan`: the real tree is built, +zero-row placeholder columns are bound for every referenced channel through +the production resolution path (a recording table — the column list cannot +drift from what binding will look up), and `.figure()` runs once — the full +mark/config validation gate (X1/X2) at compile time, in milliseconds, with +no data ingestion (constraint 2). The canonical JSON of the tree +(`plan_version: 1`) is content-addressed into a sha256-prefix `digest` and +registered in a process-local `{digest: plan}` map. Binding is the reverse: +columns + plan → a **fresh** `Chart` (never reused — X3) → `.figure()`. +Column-mismatch errors name both sides (*"plan binds column 'mag'; +Dash.cloud produced {x, y}"*). Plans refuse concrete arrays, per-mark +`data=`, and `render=` components — data-free structure only. The probe +figure also yields `dom_class_strings()`, so **live data-bound charts get +automatic Tailwind discovery** (previously live sources needed the manual +inventory). The factories that build plans, and the errors they catch at +`reflex run`, are §5. + +**Column entries.** Published columns are registry entries in their own +right, keyed by the data token: pure rebuildable caches of Reflex state +(the `@reflex_xy.data` method is the recipe), so they carry none of the +figure entry's kernel machinery — no locks (a republish replaces the whole +immutable generation and bumps its version), no pins, always sweepable +under the same TTL. The registry also keeps the `data token → {digests}` +index that lets a column republish rebuild every mounted dependent plan, +and an error seam (`on_error`) for room-wide failures that answer no +request; §4 covers the fan-out those two enable. + +**Plan format stability.** `plan_version: 1` is part of the canonical +serialization and a golden digest is pinned in +`tests/reflex_adapter/test_plan.py`. Digests are content addresses, not a +migration surface: after a format (or grammar) change, old subscribers' +digests simply miss and resync against the recompiled page — the golden +exists to catch *accidental* churn, and an intentional change bumps +`PLAN_VERSION` and re-records it. + ## 4. Updates and streaming - **State-driven rebuild** (filter changed): the figure var recomputes, @@ -718,11 +794,15 @@ demo app models. ``` python/reflex_xy/ registry.py token -> FigureEntry(figure, version, lock); TTL; - publish/push fan-out seams; append - tokens.py xyv1 token grammar; builder discovery on vars + publish/push/error fan-out seams; append; column + entries + data-token -> digest index (§3.6) + tokens.py xyv1/xyd1/xyp1 token grammar; builder discovery handles.py FigureHandle / DataHandle[S] (+ serializers): the typed values chart state vars carry vars.py @reflex_xy.figure (FigureVar: builder-tracked deps) + data_vars.py @reflex_xy.data (DataVar: columns in, handle out) + plan.py ChartPlan: zero-row probe, canonical digest, + process-local plan map, bind (§3.6) state_bridge.py token -> state_manager -> builder rebuild hook namespace.py XYNamespace: sub/unsub/msg, payload/msg/err, affinity, rebuild-on-miss, binary attachments @@ -746,9 +826,9 @@ examples/reflex/ (repo root) Reflex showcase: figure-var drilldown with whose category toggles re-bin kernel-side, §34) examples/fastapi/ (repo root) the same charts + a live 100M drilldown served from a plain FastAPI app (no committed HTML) -tests/reflex_adapter/ token/registry/var/bridge/payload-asset units, - component compile, framework contract pins - (R1/R7/R8), and a real-websocket +tests/reflex_adapter/ token/registry/var/data-var/plan/bridge/ + payload-asset units, component compile, framework + contract pins (R1/R7/R8), and a real-websocket integration suite (uvicorn + socketio client) covering payload/pick/select/affinity/rebuild/ publish-broadcast/append/unsub diff --git a/tests/reflex_adapter/test_data_var.py b/tests/reflex_adapter/test_data_var.py new file mode 100644 index 00000000..5e97bbee --- /dev/null +++ b/tests/reflex_adapter/test_data_var.py @@ -0,0 +1,219 @@ +"""The @reflex_xy.data computed var: columns in, typed handle out.""" + +from __future__ import annotations + +import asyncio +from typing import TypedDict, get_args + +import numpy as np +import pytest +import reflex as rx +from reflex_base.vars.base import AsyncComputedVar + +import reflex_xy +import xy +from reflex_xy.data_vars import AsyncDataVar, DataVar, validate_columns +from reflex_xy.handles import DataHandle +from reflex_xy.plan import build_plan +from reflex_xy.tokens import build_plan_token, builder_of, parse_token + +from .conftest import make_router_data + + +class DataSchema(TypedDict): + x: np.ndarray + y: np.ndarray + mag: np.ndarray + + +class DataDemo(rx.State): + n: int = 16 + _scale: float = 2.0 + + @reflex_xy.data + def cloud(self) -> DataSchema: + xs = np.linspace(0.0, 1.0, self.n) + return {"x": xs, "y": xs * self._scale, "mag": np.abs(xs)} + + @reflex_xy.data + def maybe(self): + if self.n < 0: + return None + return {"x": [1.0], "y": [2.0]} + + @reflex_xy.data + async def remote(self) -> DataSchema: + await asyncio.sleep(0) + xs = np.linspace(0.0, 1.0, self.n) + return {"x": xs, "y": xs, "mag": xs} + + +def hydrated_substate(client_token: str) -> DataDemo: + root = rx.State(_reflex_internal_init=True) + root.router = make_router_data(client_token) + return root.get_substate(tuple(DataDemo.get_full_name().split("."))[1:]) + + +def test_dispatch_and_schema_channel(): + """R7: the TypedDict return annotation parametrizes the var's value type; + async methods dispatch to AsyncDataVar exactly like rx.var.""" + assert isinstance(DataDemo.computed_vars["cloud"], DataVar) + assert DataDemo.cloud._var_type == DataHandle[DataSchema] + assert get_args(DataDemo.cloud._var_type) == (DataSchema,) + # untyped methods degrade to the plain handle (columns checked at runtime) + assert DataDemo.maybe._var_type is DataHandle + assert isinstance(DataDemo.computed_vars["remote"], AsyncDataVar) + assert isinstance(DataDemo.computed_vars["remote"], AsyncComputedVar) + assert DataDemo.remote._var_type == DataHandle[DataSchema] + + +def test_deps_track_the_data_method_not_the_wrapper(): + deps = DataDemo.computed_vars["cloud"]._deps(DataDemo) + assert deps == {DataDemo.get_full_name(): {"n", "_scale"}} + + +def test_evaluation_publishes_columns_and_returns_handle(_fresh_registry, client_token): + state = hydrated_substate(client_token) + handle = state.cloud + assert isinstance(handle, DataHandle) + parsed = parse_token(handle.token) + assert parsed is not None and parsed.kind == "data" + assert parsed.client_token == client_token + assert parsed.var_name == "cloud" + entry = _fresh_registry.get_columns(handle.token) + assert entry is not None + assert sorted(entry.columns) == ["mag", "x", "y"] + assert len(entry.columns["x"]) == 16 + + +def test_republish_keeps_handle_bumps_column_version(_fresh_registry, client_token): + state = hydrated_substate(client_token) + handle = state.cloud + state.n = 32 + DataDemo.computed_vars["cloud"].mark_dirty(state) + assert state.cloud == handle # stable identity, like figure vars + entry = _fresh_registry.get_columns(handle.token) + assert entry.version == 2 + assert len(entry.columns["x"]) == 32 + + +def test_pre_session_short_circuit(_fresh_registry): + root = rx.State(_reflex_internal_init=True) + state = root.get_substate(tuple(DataDemo.get_full_name().split("."))[1:]) + assert state.cloud == DataHandle("") # data method never ran + assert len(_fresh_registry._column_entries) == 0 + + +def test_none_releases_columns(_fresh_registry, client_token): + state = hydrated_substate(client_token) + token = state.maybe.token + assert _fresh_registry.get_columns(token) is not None + state.n = -1 + DataDemo.computed_vars["maybe"].mark_dirty(state) + assert state.maybe == DataHandle("") + assert _fresh_registry.get_columns(token) is None + + +def test_async_variant_awaits_and_publishes(_fresh_registry, client_token): + state = hydrated_substate(client_token) + + async def main(): + handle = await state.remote + assert _fresh_registry.get_columns(handle.token) is not None + return handle + + handle = asyncio.run(main()) + assert handle.token.startswith("xyd1|") + + +def test_underscore_name_refused(): + with pytest.raises(ValueError, match="must not start with '_'"): + + class BadData(rx.State): # noqa: F841 - definition is the assertion + @reflex_xy.data + def _hidden(self): + return None + + +def test_method_resolvable_from_class(client_token): + method = builder_of(DataDemo, "cloud") + assert method is not None + state = hydrated_substate(client_token) + assert sorted(method(state)) == ["mag", "x", "y"] + + +@pytest.mark.parametrize( + ("columns", "match"), + [ + ([1, 2, 3], "mapping"), + ({1: [1.0]}, "column names must be strings"), + ({"x": "not-values"}, "array-like"), + ({"x": {"nested": 1}}, "array-like"), + ({"x": 3.5}, "with a length"), + ({"x": [1.0, 2.0], "y": [1.0]}, "share one length"), + ], +) +def test_validate_columns_rejects_malformed(columns, match): + with pytest.raises((TypeError, ValueError), match=match): + validate_columns(columns, source="Demo.cloud") + + +def test_republish_rebuilds_and_bumps_mounted_dependents(_fresh_registry, client_token): + """The data-token -> digest index: a column republish re-binds every + mounted plan into a fresh figure generation (broadcast by publish).""" + state = hydrated_substate(client_token) + handle = state.cloud + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + composite = build_plan_token(plan.digest, handle.token) + # mount: first bind, as the namespace does on sub + entry = _fresh_registry.publish( + composite, + plan.bind(_fresh_registry.get_columns(handle.token).columns).figure(), + broadcast=False, + ) + _fresh_registry.bind_plan(handle.token, plan.digest) + assert entry.figure.traces[0].n_points == 16 + + state.n = 48 + DataDemo.computed_vars["cloud"].mark_dirty(state) + assert state.cloud == handle + rebuilt = _fresh_registry.get(composite) + assert rebuilt.version == entry.version + 1 + assert rebuilt.figure.traces[0].n_points == 48 + + +def test_republish_drops_unmounted_dependents_from_the_index(_fresh_registry, client_token): + state = hydrated_substate(client_token) + handle = state.cloud + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + _fresh_registry.bind_plan(handle.token, plan.digest) # indexed, never mounted + + state.n = 20 + DataDemo.computed_vars["cloud"].mark_dirty(state) + assert state.cloud == handle + assert _fresh_registry._digests_by_data_token.get(handle.token) is None + + +def test_release_columns_releases_dependent_figures(_fresh_registry, client_token): + state = hydrated_substate(client_token) + handle = state.maybe.token + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + composite = build_plan_token(plan.digest, handle) + columns = _fresh_registry.get_columns(handle).columns + _fresh_registry.publish(composite, plan.bind(columns).figure(), broadcast=False) + _fresh_registry.bind_plan(handle, plan.digest) + + state.n = -1 + DataDemo.computed_vars["maybe"].mark_dirty(state) + assert state.maybe == DataHandle("") + assert _fresh_registry.get(composite) is None + assert _fresh_registry.get_columns(handle) is None + + +def test_column_entries_sweep_like_figures(_fresh_registry, client_token): + state = hydrated_substate(client_token) + token = state.cloud.token + entry = _fresh_registry.get_columns(token) + dropped = _fresh_registry.sweep(now=entry.last_access + 10**9) + assert token in dropped + assert _fresh_registry.get_columns(token) is None diff --git a/tests/reflex_adapter/test_plan.py b/tests/reflex_adapter/test_plan.py new file mode 100644 index 00000000..a7a1eeb7 --- /dev/null +++ b/tests/reflex_adapter/test_plan.py @@ -0,0 +1,113 @@ +"""Chart plans: compile-time validation, content addressing, binding.""" + +from __future__ import annotations + +import numpy as np +import pytest + +import xy +from reflex_xy.plan import ( + PLAN_VERSION, + PlanBindError, + PlanError, + PlanMissError, + build_plan, + plan_of, + require_plan, +) + + +def scatter_plan(**mark_options): + return build_plan( + "scatter_chart", + (xy.scatter("x", "y", color="mag", **mark_options), xy.x_axis(label="sigma")), + {"title": "cloud"}, + ) + + +def test_plan_records_probed_columns_in_resolution_order(): + plan = scatter_plan() + assert plan.columns == ("x", "y", "mag") + + +def test_digest_is_stable_across_identical_builds(): + """Content addressing: separately constructed identical trees agree, so + every worker evaluating the same page derives the same digest (X4).""" + assert scatter_plan().digest == scatter_plan().digest + assert scatter_plan().digest != scatter_plan(opacity=0.5).digest + + +def test_digest_golden_pins_the_plan_format(): + """plan_version is part of the serialization: accidental format churn + shows up here as a digest change. An *intentional* format change bumps + PLAN_VERSION and re-records this golden (old digests are content + addresses — stale subscribers resync, nothing migrates).""" + assert PLAN_VERSION == 1 + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + assert plan.digest == "b7d0b4245b686130e37d" + + +def test_zero_row_probe_fires_the_full_validation_gate(): + with pytest.raises(ValueError, match="colormap"): + scatter_plan(colormap="virids") + with pytest.raises(ValueError, match="symbol"): + scatter_plan(symbol="marsian") + # unresolved axis-id references are figure-compile errors too (X2) + with pytest.raises(ValueError, match="axis"): + build_plan("scatter_chart", (xy.scatter("x", "y", y_axis="y2"),), {}) + + +def test_plans_refuse_concrete_arrays(): + with pytest.raises(PlanError, match="data-free"): + build_plan("scatter_chart", (xy.scatter(np.array([1.0]), np.array([2.0])),), {}) + + +def test_plans_refuse_per_mark_data(): + with pytest.raises(PlanError, match="per-mark data="): + build_plan("scatter_chart", (xy.scatter("x", "y", data={"x": [], "y": []}),), {}) + + +def test_plans_refuse_render_components(): + with pytest.raises(PlanError, match="render"): + build_plan( + "scatter_chart", + (xy.scatter("x", "y"), xy.legend(render=object())), + {}, + ) + + +def test_bind_produces_fresh_figures_per_call(): + """X3: Chart.figure() memoizes, so bind() must mint a fresh Chart — + two binds with different columns give independent figures.""" + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + small = plan.bind({"x": [1.0], "y": [2.0]}).figure() + large = plan.bind({"x": [1.0, 2.0, 3.0], "y": [2.0, 3.0, 4.0]}).figure() + assert small is not large + assert small.traces[0].n_points == 1 + assert large.traces[0].n_points == 3 + + +def test_bind_error_names_both_sides(): + plan = scatter_plan() + with pytest.raises( + PlanBindError, match=r"plan binds column 'mag'; Dash.cloud produced \{x, y\}" + ): + plan.bind({"x": [1.0], "y": [2.0]}, source="Dash.cloud") + + +def test_probe_collects_the_tailwind_inventory(): + plan = build_plan( + "scatter_chart", + (xy.scatter("x", "y"), xy.legend(class_name="max-h-24 overflow-y-auto")), + {"class_name": "rounded-xl"}, + ) + assert "rounded-xl" in plan.tailwind_classes + assert "max-h-24" in plan.tailwind_classes + + +def test_registry_lookup_and_miss(): + plan = scatter_plan() + assert plan_of(plan.digest) is plan + assert plan_of("feedfacefeedfacefeed") is None + with pytest.raises(PlanMissError, match="feedfacefeedfacefeed"): + require_plan("feedfacefeedfacefeed") From e3f4b13d4e5c5c561be45988121585b874ee6de2 Mon Sep 17 00:00:00 2001 From: Farhan Ali Raza Date: Thu, 6 Aug 2026 17:26:10 +0500 Subject: [PATCH 2/3] fix(reflex): generation-gated column republish and snapshot plans Two review P-findings on the plan/data tier: - publish_columns rebuilds run outside the registry mutex, so two republishes of one data token could finish in reverse order and leave subscribers on stale pixels. Each rebuild now carries its ColumnEntry generation and every outcome (figure publish, failure release + err frame) is gated on that generation still being current, atomically with the registry mutation (publish's locked body factored into _publish_locked for that). - build_plan hashes and registers a deep snapshot of children and chart props: mutating a reused mark node after the factory call can no longer change binding behavior behind an unchanged digest. --- python/reflex_xy/plan.py | 7 ++ python/reflex_xy/registry.py | 111 +++++++++++++++++--------- spec/design/reflex-integration.md | 18 +++++ tests/reflex_adapter/test_data_var.py | 35 ++++++++ tests/reflex_adapter/test_plan.py | 19 +++++ 5 files changed, 151 insertions(+), 39 deletions(-) diff --git a/python/reflex_xy/plan.py b/python/reflex_xy/plan.py index 4e828924..28f54628 100644 --- a/python/reflex_xy/plan.py +++ b/python/reflex_xy/plan.py @@ -28,6 +28,7 @@ from __future__ import annotations +import copy import dataclasses import hashlib import json @@ -163,6 +164,12 @@ def build_plan( "chart-level data source (this is tracked as deferred work in " "spec/design/reflex-component-api-implementation.md)" ) + # The plan must be immutable once addressed: hash a deep snapshot and + # register *that* snapshot, so mutating a reused mark node or a props + # dict after the factory call can never change binding behavior behind + # an unchanged digest/columns record. + children = copy.deepcopy(children) + chart_props = copy.deepcopy(chart_props) serialized = { "plan_version": PLAN_VERSION, "kind": kind, diff --git a/python/reflex_xy/registry.py b/python/reflex_xy/registry.py index 3c394246..6ea57d8b 100644 --- a/python/reflex_xy/registry.py +++ b/python/reflex_xy/registry.py @@ -334,38 +334,7 @@ def publish( which is the signal that subscribers need a new payload. """ with self._mutex: - entry = self._entries.get(token) - # A cache-miss rebuild inserts its entry before room fan-out. A - # canonical same-object publish re-authorizes that provisional - # generation without bumping its version, but must still own a - # normal broadcast in case the rebuild-owned fan-out fails. - reauthorized = entry is not None and bool(self._active_rebuild_guards.get(token)) - # This is a canonical dependency/application publish. Any builder - # that began from the preceding absence must not overwrite it, - # even if this entry is released again before that builder ends. - self._invalidate_rebuild_guards_locked(token) - if entry is None: - version = self._evicted_versions.pop(token, 0) + 1 - entry = FigureEntry(figure=figure, token=token, version=version, pinned=pinned) - self._entries[token] = entry - changed = True - else: - changed = entry.figure is not figure - if changed: - # A replacement is a new immutable generation. In-flight - # handlers keep a self-consistent old figure/version pair - # instead of observing the old figure with a newly bumped - # version. Their results are discarded by ``is_current``. - entry = FigureEntry( - figure=figure, - token=token, - version=entry.version + 1, - pinned=entry.pinned or pinned, - ) - self._entries[token] = entry - else: - entry.pinned = entry.pinned or pinned - entry.touch() + entry, changed, reauthorized = self._publish_locked(token, figure, pinned) if broadcast and (changed or reauthorized): # Re-publishing the identical object means nothing moved; a new # figure object is normally the signal subscribers need a fresh @@ -375,6 +344,48 @@ def publish( self.schedule_broadcast(token) return entry + def _publish_locked( + self, token: str, figure: "Figure", pinned: bool + ) -> tuple[FigureEntry, bool, bool]: + """The mutation half of ``publish``; the registry mutex must be held. + + Returns ``(entry, changed, reauthorized)`` so callers can schedule the + broadcast after releasing the mutex. + """ + entry = self._entries.get(token) + # A cache-miss rebuild inserts its entry before room fan-out. A + # canonical same-object publish re-authorizes that provisional + # generation without bumping its version, but must still own a + # normal broadcast in case the rebuild-owned fan-out fails. + reauthorized = entry is not None and bool(self._active_rebuild_guards.get(token)) + # This is a canonical dependency/application publish. Any builder + # that began from the preceding absence must not overwrite it, + # even if this entry is released again before that builder ends. + self._invalidate_rebuild_guards_locked(token) + if entry is None: + version = self._evicted_versions.pop(token, 0) + 1 + entry = FigureEntry(figure=figure, token=token, version=version, pinned=pinned) + self._entries[token] = entry + changed = True + else: + changed = entry.figure is not figure + if changed: + # A replacement is a new immutable generation. In-flight + # handlers keep a self-consistent old figure/version pair + # instead of observing the old figure with a newly bumped + # version. Their results are discarded by ``is_current``. + entry = FigureEntry( + figure=figure, + token=token, + version=entry.version + 1, + pinned=entry.pinned or pinned, + ) + self._entries[token] = entry + else: + entry.pinned = entry.pinned or pinned + entry.touch() + return entry, changed, reauthorized + def begin_rebuild(self, token: str) -> tuple[Optional[FigureEntry], Optional[object]]: """Atomically return the current entry or guard this missing generation.""" with self._mutex: @@ -549,7 +560,7 @@ def publish_columns(self, token: str, columns: dict[str, Any]) -> ColumnEntry: self._column_entries[token] = entry digests = sorted(self._digests_by_data_token.get(token, ())) for digest in digests: - self._rebuild_dependent(token, digest, columns) + self._rebuild_dependent(token, digest, entry) return entry def get_columns(self, token: str) -> Optional[ColumnEntry]: @@ -580,13 +591,23 @@ def bind_plan(self, data_token: str, digest: str) -> None: with self._mutex: self._digests_by_data_token.setdefault(data_token, set()).add(digest) - def _rebuild_dependent(self, data_token: str, digest: str, columns: dict[str, Any]) -> None: - """Re-bind one dependent plan against freshly published columns.""" + def _rebuild_dependent(self, data_token: str, digest: str, column_entry: ColumnEntry) -> None: + """Re-bind one dependent plan against freshly published columns. + + ``column_entry`` is the generation this rebuild serves. Concurrent + republishes rebuild concurrently and may complete out of order; every + outcome below (publish, failure release + err frame) is gated on the + generation still being current — atomically with the registry mutation + — so an older generation that finishes late can never overwrite a + newer one's figure or report a stale failure. + """ from .plan import plan_of from .tokens import build_plan_token, parse_token composite = build_plan_token(digest, data_token) with self._mutex: + if self._column_entries.get(data_token) is not column_entry: + return # a newer generation owns this rebuild now mounted = composite in self._entries or bool( self._rebuildable_subscribers.get(composite) ) @@ -610,16 +631,28 @@ def _rebuild_dependent(self, data_token: str, digest: str, columns: dict[str, An from .plan import PlanMissError raise PlanMissError(digest) - figure = plan.bind(columns, source=source).figure() + figure = plan.bind(column_entry.columns, source=source).figure() except Exception as exc: # noqa: BLE001 - user data meets page structure here # Fail loud on both sides: the server log carries the bind error, # and subscribers get an err frame instead of silently frozen - # pixels (their bounded resync retries against current state). + # pixels (their bounded resync retries against current state) — + # unless a newer generation superseded this one meanwhile, in + # which case the failure is obsolete and it owns the outcome. + with self._mutex: + if self._column_entries.get(data_token) is not column_entry: + return + self._invalidate_rebuild_guards_locked(composite) + removed = self._entries.pop(composite, None) + self._retain_removed_version_locked(composite, removed) _logger.warning("republish of %s failed: %s", composite, exc) - self.release(composite) self._schedule_error(composite, str(exc), resync=True) return - self.publish(composite, figure) + with self._mutex: + if self._column_entries.get(data_token) is not column_entry: + return # a newer generation's figure must win; drop this one + _, changed, reauthorized = self._publish_locked(composite, figure, False) + if changed or reauthorized: + self.schedule_broadcast(composite) # -- version bump + fan-out --------------------------------------------- diff --git a/spec/design/reflex-integration.md b/spec/design/reflex-integration.md index fe9927f1..4f97b6cf 100644 --- a/spec/design/reflex-integration.md +++ b/spec/design/reflex-integration.md @@ -493,6 +493,24 @@ index that lets a column republish rebuild every mounted dependent plan, and an error seam (`on_error`) for room-wide failures that answer no request; §4 covers the fan-out those two enable. +**Republish ordering.** Dependent rebuilds run outside the registry mutex +(they execute user-scale figure builds), so two republishes of one data +token may finish in reverse order. Every rebuild therefore carries the +`ColumnEntry` generation it serves, and each of its outcomes — the figure +publish, and the failure path's release + `err {resync}` frame — is gated +on that generation still being current, *atomically with the registry +mutation*. An older generation that finishes late is dropped whole: +subscribers can never observe a newer generation's pixels replaced by an +older one's, and a superseded failure raises no stale error. Pinned by +`test_data_var.py::test_stale_column_generation_cannot_overwrite_a_newer_publish`. + +**Plan immutability.** `build_plan` hashes — and registers — a deep +snapshot of the node tree and chart props, not the caller's objects. +Page code that reuses and later mutates a mark node (or the props dict) +cannot change binding behavior behind an unchanged digest/columns record; +the canonical representation is the source of truth. Pinned by +`test_plan.py::test_plan_is_a_snapshot_immune_to_later_node_mutation`. + **Plan format stability.** `plan_version: 1` is part of the canonical serialization and a golden digest is pinned in `tests/reflex_adapter/test_plan.py`. Digests are content addresses, not a diff --git a/tests/reflex_adapter/test_data_var.py b/tests/reflex_adapter/test_data_var.py index 5e97bbee..20698d4e 100644 --- a/tests/reflex_adapter/test_data_var.py +++ b/tests/reflex_adapter/test_data_var.py @@ -182,6 +182,41 @@ def test_republish_rebuilds_and_bumps_mounted_dependents(_fresh_registry, client assert rebuilt.figure.traces[0].n_points == 48 +def test_stale_column_generation_cannot_overwrite_a_newer_publish(_fresh_registry, client_token): + """Two concurrent republishes may finish their rebuilds in reverse order; + the older generation's late publish must be dropped, atomically with the + currency check, so subscribers never regress to stale pixels.""" + state = hydrated_substate(client_token) + handle = state.cloud + plan = build_plan("scatter_chart", (xy.scatter("x", "y"),), {}) + composite = build_plan_token(plan.digest, handle.token) + _fresh_registry.publish( + composite, + plan.bind(_fresh_registry.get_columns(handle.token).columns).figure(), + broadcast=False, + ) + _fresh_registry.bind_plan(handle.token, plan.digest) + + old_generation = _fresh_registry.get_columns(handle.token) + state.n = 48 + DataDemo.computed_vars["cloud"].mark_dirty(state) + assert state.cloud == handle + current = _fresh_registry.get(composite) + assert current.figure.traces[0].n_points == 48 + + # The older generation's rebuild finishes late: its publish must be a + # no-op (same for its failure path — no release, no stale err frame). + _fresh_registry._rebuild_dependent(handle.token, plan.digest, old_generation) + after = _fresh_registry.get(composite) + assert after is current + assert after.figure.traces[0].n_points == 48 + assert after.version == current.version + + broken = type(old_generation)(columns={"x": [1.0]}, token=handle.token, version=1) + _fresh_registry._rebuild_dependent(handle.token, plan.digest, broken) + assert _fresh_registry.get(composite) is current # stale failure: no release + + def test_republish_drops_unmounted_dependents_from_the_index(_fresh_registry, client_token): state = hydrated_substate(client_token) handle = state.cloud diff --git a/tests/reflex_adapter/test_plan.py b/tests/reflex_adapter/test_plan.py index a7a1eeb7..22bddae2 100644 --- a/tests/reflex_adapter/test_plan.py +++ b/tests/reflex_adapter/test_plan.py @@ -76,6 +76,25 @@ def test_plans_refuse_render_components(): ) +def test_plan_is_a_snapshot_immune_to_later_node_mutation(): + """The registered plan holds a deep copy of what was hashed: mutating a + reused mark node or the chart props afterwards must not change binding + behavior behind an unchanged digest/columns record.""" + mark = xy.scatter("x", "y", color="mag") + props = {"title": "cloud"} + plan = build_plan("scatter_chart", (mark,), props) + assert plan.columns == ("x", "y", "mag") + + mark.props["color"] = "sneaky" # node reused and mutated by page code + props["title"] = "renamed" + assert plan.children[0] is not mark + assert plan.children[0].props["color"] == "mag" + assert plan.chart_props == {"title": "cloud"} + # binding still resolves the snapshot's channels, not the mutated node's + fig = plan.bind({"x": [1.0], "y": [2.0], "mag": [3.0]}).figure() + assert fig.traces[0].n_points == 1 + + def test_bind_produces_fresh_figures_per_call(): """X3: Chart.figure() memoizes, so bind() must mint a fresh Chart — two binds with different columns give independent figures.""" From 794dae62b2c180d0248877ecd4eae89056e9c48a Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Thu, 6 Aug 2026 22:00:11 +0000 Subject: [PATCH 3/3] refactor(reflex): share the builder-target dep helper between var tiers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit data_vars.py duplicated vars.py's _builder_target verbatim; a fix landing in one copy would silently miss the other. The data tier now imports the figure tier's helper — one body for the one behavior, matching the 'exact sibling' contract the module documents. --- python/reflex_xy/data_vars.py | 9 +-------- python/reflex_xy/vars.py | 5 +++-- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/python/reflex_xy/data_vars.py b/python/reflex_xy/data_vars.py index 2dd72fa7..0ba1573b 100644 --- a/python/reflex_xy/data_vars.py +++ b/python/reflex_xy/data_vars.py @@ -33,18 +33,11 @@ from .handles import DataHandle from .registry import registry from .tokens import BUILDER_ATTR, build_data_token +from .vars import _builder_target __all__ = ["AsyncDataVar", "DataVar", "data", "validate_columns"] -def _builder_target(var: Any, obj: Any) -> Any: - """Point dependency tracking at the *data method*, not the token wrapper - (same reason as vars.py: the wrapper fget reads nothing but the router).""" - if obj is not None: - return obj - return getattr(var._fget, BUILDER_ATTR, None) - - class DataVar(ComputedVar): """ComputedVar whose value is a DataHandle (sync data method).""" diff --git a/python/reflex_xy/vars.py b/python/reflex_xy/vars.py index 53f7190c..4898956f 100644 --- a/python/reflex_xy/vars.py +++ b/python/reflex_xy/vars.py @@ -46,8 +46,9 @@ def _builder_target(var: Any, obj: Any) -> Any: - """Point dependency tracking at the *builder*, not the token wrapper: - reflex should track what the chart reads, and the wrapper fget reads + """Point dependency tracking at the wrapped method — the figure builder + here, the data method in data_vars.py — not the token wrapper: reflex + should track what the method body reads, and the wrapper fget reads nothing but the router.""" if obj is not None: return obj