perf: split router into per-field base vars; gather connection-static router data at connect

news/7068.deprecation.md

@@ -0,0 +1 @@
+Declaring a computed var dependency on the `router` var (`deps=["router"]`) is deprecated; depend on the specific router var instead, e.g. `deps=["router_url"]`.

news/7068.performance.md

@@ -0,0 +1 @@
+Store router data in separate base vars (session, headers, page, url, route_id) so a navigation delta only re-sends the fields that changed instead of the whole router, and gather the connection-scoped router data (headers, client IP, session id) once at connect time rather than on every event. `State.router` is unchanged for app code.

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

@@ -0,0 +1 @@
+The event processor now refreshes only the router vars whose backing `router_data` keys actually changed, so a navigation no longer rebuilds and re-sends the connection-scoped session and header data. `ROUTER_VARS` names the per-field router vars that replaced the single `router` var on the root state.

packages/reflex-base/src/reflex_base/constants/__init__.py

@@ -60,6 +60,12 @@
     ROUTER,
     ROUTER_DATA,
     ROUTER_DATA_INCLUDE,
+    ROUTER_HEADERS,
+    ROUTER_PAGE,
+    ROUTER_ROUTE_ID,
+    ROUTER_SESSION,
+    ROUTER_URL,
+    ROUTER_VARS,
     DefaultPage,
     Page404,
     RouteArgType,
@@ -86,6 +92,12 @@
     "ROUTER",
     "ROUTER_DATA",
     "ROUTER_DATA_INCLUDE",
+    "ROUTER_HEADERS",
+    "ROUTER_PAGE",
+    "ROUTER_ROUTE_ID",
+    "ROUTER_SESSION",
+    "ROUTER_URL",
+    "ROUTER_VARS",
     "ROUTE_NOT_FOUND",
     "SESSION_STORAGE",
     "SETTER_PREFIX",

packages/reflex-base/src/reflex_base/constants/route.py

@@ -11,10 +11,28 @@ class RouteArgType(SimpleNamespace):
     LIST = "arg_list"
 
 
-# the name of the backend var containing path and client information
+# the name of the state attribute exposing path and client information
 ROUTER = "router"
 ROUTER_DATA = "router_data"
 
+# The names of the per-field base vars holding router data on the root state.
+# Session and headers are constant for the lifetime of a websocket connection,
+# while page, url, and route_id change on every navigation; keeping them in
+# separate vars means a navigation delta only re-sends the navigation fields.
+ROUTER_SESSION = "router_session"
+ROUTER_HEADERS = "router_headers"
+ROUTER_PAGE = "router_page"
+ROUTER_URL = "router_url"
+ROUTER_ROUTE_ID = "router_route_id"
+
+ROUTER_VARS = (
+    ROUTER_SESSION,
+    ROUTER_HEADERS,
+    ROUTER_PAGE,
+    ROUTER_URL,
+    ROUTER_ROUTE_ID,
+)
+
 
 class RouteVar(SimpleNamespace):
     """Names of variables used in the router_data dict stored in State."""

packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py

@@ -12,7 +12,6 @@
 from importlib.util import find_spec
 from typing import TYPE_CHECKING, Any
 
-from reflex.istate.data import RouterData
 from reflex.istate.manager.token import BaseStateToken
 from reflex.istate.proxy import StateProxy
 from reflex.utils import types
@@ -414,12 +413,16 @@ async def _execute_event(
             )
 
             # re-assign only when the value is set and different
-            if router_data and state.router_data != router_data:
+            if (
+                router_data
+                and (previous_router_data := state.router_data) != router_data
+            ):
                 # assignment will recurse into substates and force recalculation of
                 # dependent ComputedVar (dynamic route variables)
                 state.router_data = router_data
-                if state.router != (router := RouterData.from_router_data(router_data)):
-                    state.router = router
+                # only the router vars whose backing keys changed are rebuilt
+                # and re-sent; session/headers stay put across navigations
+                state._update_router_vars(router_data, previous_router_data)
 
             # Preprocess the event.
             if (

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

@@ -720,6 +720,21 @@ def _get_all_var_data(self) -> VarData | None:
         """
         return self._var_data
 
+    def _dependency_field_names(self) -> tuple[str, ...]:
+        """The state field names a ComputedVar depending on this Var must track.
+
+        A Var normally stands for a single state field, the one named by its
+        VarData. A Var composed of several state fields must name all of them,
+        or a ``deps=[that_var]`` dependency would only track the one field
+        VarData.merge happened to surface, leaving the computed var stale when
+        any of the others change.
+
+        Returns:
+            The field names to register the dependency against.
+        """
+        all_var_data = self._get_all_var_data()
+        return (all_var_data.field_name if all_var_data is not None else "",)
+
     def __deepcopy__(self, memo: dict[int, Any]) -> Self:
         """Deepcopy the var.
 
@@ -2411,10 +2426,11 @@ def _add_static_dep(
                 else None
             )
             if all_var_data is not None:
-                var_name = all_var_data.field_name
+                # A composite Var names every state field it is built from.
+                var_names = dep._dependency_field_names()
             else:
-                var_name = dep._js_expr
-            deps.setdefault(state_name, set()).add(var_name)
+                var_names = (dep._js_expr,)
+            deps.setdefault(state_name, set()).update(var_names)
         elif isinstance(dep, str) and dep != "":
             deps.setdefault(None, set()).add(dep)
         else:
@@ -2692,18 +2708,20 @@ def add_dependency(self, objclass: type[BaseState], dep: Var):
         if all_var_data := dep._get_all_var_data():
             state_name = all_var_data.state
             if state_name:
-                var_name = all_var_data.field_name
-                if var_name:
-                    self._static_deps.setdefault(state_name, set()).add(var_name)
+                # A composite Var names every state field it is built from.
+                var_names = tuple(filter(None, dep._dependency_field_names()))
+                if var_names:
+                    self._static_deps.setdefault(state_name, set()).update(var_names)
                     target_state_class = objclass.get_root_state().get_class_substate(
                         state_name
                     )
-                    target_state_class._var_dependencies.setdefault(
-                        var_name, set()
-                    ).add((
-                        objclass.get_full_name(),
-                        self._name,
-                    ))
+                    for var_name in var_names:
+                        target_state_class._var_dependencies.setdefault(
+                            var_name, set()
+                        ).add((
+                            objclass.get_full_name(),
+                            self._name,
+                        ))
                     target_state_class._potentially_dirty_states.add(
                         objclass.get_full_name()
                     )

reflex/app.py

@@ -73,7 +73,7 @@
 from reflex.app_mixins import AppMixin, LifespanMixin, MiddlewareMixin
 from reflex.compiler import compiler
 from reflex.compiler.compiler import readable_name_from_component
-from reflex.istate.data import RouterData
+from reflex.istate.data import SessionData
 from reflex.istate.manager import StateManager, StateModificationContext
 from reflex.istate.manager.token import BaseStateToken
 from reflex.route import (
@@ -1977,6 +1977,10 @@ def __init__(self, namespace: str, app: App):
         # Number of client_error reports logged per SID, for rate limiting.
         self._client_error_counts: dict[str, int] = {}
 
+        # Connection-scoped router_data entries per SID, computed once at
+        # connect time instead of for every event on the connection.
+        self._static_router_data: dict[str, dict[str, Any]] = {}
+
         # Start time and count of the current process-wide client_error window.
         self._client_error_window_start = 0.0
         self._client_error_window_count = 0
@@ -2026,6 +2030,51 @@ async def on_connect(self, sid: str, environ: dict):
                 f"Frontend version {subprotocol} for session {sid} does not match the backend version {constants.Reflex.VERSION}."
             )
 
+        # Headers, client IP, and session id cannot change for the lifetime of
+        # the connection; compute them once instead of on every event.
+        self._static_router_data[sid] = self._build_static_router_data(sid, environ)
+
+    def _build_static_router_data(self, sid: str, environ: dict) -> dict[str, Any]:
+        """Build the connection-scoped router_data entries for a socket.
+
+        Args:
+            sid: The Socket.IO session id.
+            environ: The request information, including HTTP headers.
+
+        Returns:
+            The router_data entries that are constant for the connection.
+        """
+        asgi_scope = environ.get("asgi.scope", {})
+
+        # Get the client headers.
+        headers = {
+            k.decode("utf-8"): v.decode("utf-8")
+            for (k, v) in asgi_scope.get("headers", [])
+        }
+
+        # Get the client IP
+        try:
+            client_ip = asgi_scope["client"][0]
+            headers["asgi-scope-client"] = client_ip
+        except (KeyError, IndexError):
+            client_ip = environ.get("REMOTE_ADDR", "0.0.0.0")
+
+        # Unroll reverse proxy forwarded headers.
+        client_ip = (
+            headers
+            .get(
+                "x-forwarded-for",
+                client_ip,
+            )
+            .partition(",")[0]
+            .strip()
+        )
+        return {
+            constants.RouteVar.SESSION_ID: sid,
+            constants.RouteVar.HEADERS: headers,
+            constants.RouteVar.CLIENT_IP: client_ip,
+        }
+
     def on_disconnect(self, sid: str) -> asyncio.Task | None:
         """Event for when the websocket disconnects.
 
@@ -2036,6 +2085,7 @@ def on_disconnect(self, sid: str) -> asyncio.Task | None:
             An asyncio Task for cleaning up the token, or None.
         """
         self._client_error_counts.pop(sid, None)
+        self._static_router_data.pop(sid, None)
         # Get token before cleaning up
         disconnect_token = self.sid_to_token.get(sid)
         if disconnect_token:
@@ -2128,45 +2178,33 @@ async def on_event(self, sid: str, data: Any):
             msg = f"Failed to deserialize event data: {fields}."
             raise exceptions.EventDeserializationError(msg) from ex
 
-        # Get the event environment.
-        if self.app.sio is None:
-            msg = "Socket.IO is not initialized."
-            raise RuntimeError(msg)
-        environ = self.app.sio.get_environ(sid, self.namespace)
-        if environ is None:
-            msg = "Socket.IO environ is not initialized."
-            raise RuntimeError(msg)
-
-        # Get the client headers.
-        headers = {
-            k.decode("utf-8"): v.decode("utf-8")
-            for (k, v) in environ["asgi.scope"]["headers"]
-        }
-
-        # Get the client IP
-        try:
-            client_ip = environ["asgi.scope"]["client"][0]
-            headers["asgi-scope-client"] = client_ip
-        except (KeyError, IndexError):
-            client_ip = environ.get("REMOTE_ADDR", "0.0.0.0")
-
-        # Unroll reverse proxy forwarded headers.
-        client_ip = (
-            headers
-            .get(
-                "x-forwarded-for",
-                client_ip,
+        static_router_data = self._static_router_data.get(sid)
+        if static_router_data is None:
+            # The connection was not seen by on_connect (e.g. namespace created
+            # after the socket connected); fall back to the connection environ.
+            if self.app.sio is None:
+                msg = "Socket.IO is not initialized."
+                raise RuntimeError(msg)
+            environ = self.app.sio.get_environ(sid, self.namespace)
+            if environ is None:
+                msg = "Socket.IO environ is not initialized."
+                raise RuntimeError(msg)
+            static_router_data = self._static_router_data[sid] = (
+                self._build_static_router_data(sid, environ)
             )
-            .partition(",")[0]
-            .strip()
-        )
         router_data = event.router_data
+        router_data.update(static_router_data)
+        # The cached headers reach the event, and from there `state.router_data`,
+        # which is a plain mutable dict: sharing the mapping would let a handler
+        # mutating `self.router_data["headers"]` corrupt the connection cache for
+        # every later event on this socket. The shallow copy is ~17x cheaper than
+        # the per-event header decode it replaced, so the cache still pays off.
+        router_data[constants.RouteVar.HEADERS] = static_router_data[
+            constants.RouteVar.HEADERS
+        ].copy()
         router_data.update({
             constants.RouteVar.QUERY: format.format_query_params(event.router_data),
             constants.RouteVar.CLIENT_TOKEN: token,
-            constants.RouteVar.SESSION_ID: sid,
-            constants.RouteVar.HEADERS: headers,
-            constants.RouteVar.CLIENT_IP: client_ip,
         })
         router_data[constants.RouteVar.PATH] = "/" + (
             self.app.router(path) or "404"
@@ -2284,4 +2322,10 @@ async def link_token_to_sid(self, sid: str, token: str):
                 BaseStateToken(ident=new_token or token, cls=self.app._state)
             ) as state:
                 state.router_data[constants.RouteVar.SESSION_ID] = sid
-                state.router = RouterData.from_router_data(state.router_data)
+                # Rebuild from router_data (rather than replacing the field on
+                # the existing value) to keep the session var and router_data
+                # in step, the same way the event processor refreshes it.
+                if (
+                    session := SessionData.from_router_data(state.router_data)
+                ) != state.router_session:
+                    state.router_session = session