feat(server): stats_request, stats_update and df_meta.stats on deferred sessions (rows-first s3)

buckaroo/dataflow/dataflow.py

@@ -404,8 +404,10 @@
         # Recorded once the stats exist: after a run that raises, the key still
         # names the frame summary_sd belongs to, which _populate_sd_cache checks.
         self._summary_sd_cache_key = key
-        self.summary_sd = result_summary_sd
+        # errs first: assigning summary_sd runs _populate_sd_cache, which files
+        # them with the sd it caches (see _summary_errs_cache).
         self.errs = errs
+        self.summary_sd = result_summary_sd
 
     @observe('summary_sd', 'processed_result')
     @exception_protect('merged_sd-protector')
@@ -465,6 +467,9 @@
             raise ValueError(f"stats_tier must be one of {STATS_TIERS}, got {stats_tier!r}")
         # Set before super().__init__ — assigning raw_df runs the whole cascade.
         self.stats_tier = stats_tier
+        # The errs of the runs behind summary_stats_cache entries that had any,
+        # by the same key (see set_stats_tier).
+        self._summary_errs_cache: TDict[str, ErrDict] = {}
         self.init_sd: InitSD
         if init_sd is None:
             self.init_sd = {}
@@ -708,6 +713,8 @@
         current = (id(self.processed_df), id(self.analysis_klasses), self.stats_tier)
         if keys['filt'] not in new_cache and self._summary_sd_cache_key == current:
             new_cache[keys['filt']] = dict(self.summary_sd or {})
+            if getattr(self, 'errs', None):
+                self._summary_errs_cache[keys['filt']] = self.errs
             cache_grew = True
 
         # raw + clean: fresh compute, but only on cache miss.
@@ -811,23 +818,35 @@
         clean scopes from the cache, computing the ones it lacks.
         """
         filt_key = self._scope_cache_key(split_chain_by_scope(self.operations)['filt'], tier=tier)
-        if summary is None:
-            cached = self.summary_stats_cache.get(filt_key)
-            if cached is None:
-                self.stats_tier = tier
-                return
-            summary = (cached, {})
-        else:
-            # Stored under the state's key at the new tier, so merged_sd reads
-            # the stats given rather than an entry already there.
-            self.summary_stats_cache = {**self.summary_stats_cache, filt_key: summary[0]}
-        sd, errs = summary
-        # Recorded first, so the stats_tier observer finds this frame's stats
-        # present and leaves them to the assignments below.
-        self._summary_sd_cache_key = (id(self.processed_df), id(self.analysis_klasses), tier)
-        self.stats_tier = tier
-        self.summary_sd = sd
-        self.errs = errs
+        prior_tier, prior_key = self.stats_tier, self._summary_sd_cache_key
+        try:
+            if summary is None:
+                cached = self.summary_stats_cache.get(filt_key)
+                if cached is None:
+                    self.stats_tier = tier
+                    return
+                # The errs of the run that made the entry, kept apart because
+                # the cache holds sds only.
+                summary = (cached, self._summary_errs_cache.get(filt_key, {}))
+            else:
+                # Stored under the state's key at the new tier, so merged_sd reads
+                # the stats given rather than an entry already there.
+                self.summary_stats_cache = {**self.summary_stats_cache, filt_key: summary[0]}
+                if summary[1]:
+                    self._summary_errs_cache[filt_key] = summary[1]
+            sd, errs = summary
+            # Recorded first, so the stats_tier observer finds this frame's stats
+            # present and leaves them to the assignments below.
+            self._summary_sd_cache_key = (id(self.processed_df), id(self.analysis_klasses), tier)
+            self.stats_tier = tier
+            self.errs = errs
+            self.summary_sd = sd
+        except Exception:
+            # The key goes back first, so the observer the tier change fires
+            # finds the frame's stats present and runs nothing.
+            self._summary_sd_cache_key = prior_key
+            self.stats_tier = prior_tier
+            raise
 
     # ### end summary stats block
 

buckaroo/server/handlers.py

@@ -16,7 +16,8 @@
 from buckaroo.server.focus import find_or_create_session_window
 from buckaroo.dataflow.dataflow import STATS_TIERS
 from buckaroo.server.session import (
-    DEFAULT_STATS_DELIVERY, DEFAULT_STATS_TIER, STATS_DELIVERIES, build_state_message, dataflow_stats_tier)
+    DEFAULT_STATS_DELIVERY, DEFAULT_STATS_TIER, STATS_DELIVERIES, begin_stats_generation, dataflow_stats_tier)
+from buckaroo.server.stats_wire import broadcast_state, refresh_session_snapshot
 from buckaroo.server import telemetry
 from buckaroo.pluggable_analysis_framework import perf_log
 
@@ -194,17 +195,10 @@ def _push_state_to_clients(self, session, metadata: dict):
         if not session.ws_clients:
             return
 
-        for client in list(session.ws_clients):
-            try:
-                # Reset per-client live search first (dataset changed),
-                # then build the msg so the injected ``buckaroo_state``
-                # mirrors the post-reset value.
-                client.search_string = ""
-                msg = build_state_message(session, metadata=metadata,
-                    search_string=client.search_string)
-                client.write_message(json.dumps(msg))
-            except Exception:
-                session.ws_clients.discard(client)
+        # Each client's search is reset first (dataset changed), then its
+        # message is built so the injected ``buckaroo_state`` mirrors the
+        # post-reset value.
+        broadcast_state(session, metadata=metadata, reset_search=True)
 
     def _handle_browser_window(self, session_id: str) -> str:
         """Handle browser window management."""
@@ -391,6 +385,11 @@ async def post(self):
                         **component_config,
                     }
 
+        # /load builds no deferred stats, whatever policy a prior /load_expr on
+        # this session left behind, and the generation moves on with the data.
+        session.stats_tier, session.stats_delivery = "full", "inline"
+        begin_stats_generation(session)
+
         # Notify connected clients and open browser
         self._push_state_to_clients(session, metadata)
         browser_action = "skipped" if no_browser else self._handle_browser_window(session_id)
@@ -670,17 +669,13 @@ async def post(self):
                         **dvc.get("component_config", {}),
                         **component_config}
 
-        if session.ws_clients:
-            for client in list(session.ws_clients):
-                try:
-                    # Reset per-client live search (#851): a term from
-                    # the prior expression would silently filter the new one.
-                    client.search_string = ""
-                    msg = build_state_message(session, metadata=metadata,
-                        search_string=client.search_string)
-                    client.write_message(json.dumps(msg))
-                except Exception:
-                    session.ws_clients.discard(client)
+        # A new expression is a new stats generation; a deferred session starts
+        # it pending.
+        begin_stats_generation(session)
+
+        # Reset per-client live search (#851): a term from the prior
+        # expression would silently filter the new one.
+        broadcast_state(session, metadata=metadata, reset_search=True)
 
         if no_browser or not self.application.settings.get("open_browser", False):
             browser_action = "skipped"
@@ -824,18 +819,12 @@ async def post(self):
         session.df_meta = display_state["df_meta"]
         session.mode = "viewer"
         telemetry.arm_session(session, tele_sink)
+        # A viewer session has no dataflow and so no deferred stats.
+        session.stats_tier, session.stats_delivery = DEFAULT_STATS_TIER, DEFAULT_STATS_DELIVERY
+        begin_stats_generation(session)
 
-        # Push to WebSocket clients
-        if session.ws_clients:
-            for client in list(session.ws_clients):
-                try:
-                    # Reset per-client live search (#851).
-                    client.search_string = ""
-                    msg = build_state_message(session,
-                        search_string=client.search_string)
-                    client.write_message(json.dumps(msg))
-                except Exception:
-                    session.ws_clients.discard(client)
+        # Push to WebSocket clients. Reset per-client live search (#851).
+        broadcast_state(session, reset_search=True)
 
         # Browser window
         if no_browser or not self.application.settings.get("open_browser", False):
@@ -953,31 +942,13 @@ async def post(self, session_id):
         if bs.get("quick_command_args"):
             xorq_dataflow.quick_command_args = bs["quick_command_args"]
 
-        refreshed = get_buckaroo_display_state(xorq_dataflow)
         session.xorq_dataflow = xorq_dataflow
         session.stats_tier = stats_tier
         session.stats_delivery = stats_delivery
-        session.df_display_args = refreshed["df_display_args"]
-        session.df_data_dict = refreshed["df_data_dict"]
-        session.df_meta = refreshed["df_meta"]
-        session.buckaroo_options = refreshed["buckaroo_options"]
-        session.command_config = refreshed["command_config"]
+        refresh_session_snapshot(session, xorq_dataflow)
+        begin_stats_generation(session)
 
-        if session.component_config and session.df_display_args:
-            for key in session.df_display_args:
-                dvc = session.df_display_args[key].get("df_viewer_config")
-                if dvc is not None:
-                    dvc["component_config"] = {
-                        **dvc.get("component_config", {}),
-                        **session.component_config}
-
-        for client in list(session.ws_clients):
-            try:
-                msg = build_state_message(session,
-                    search_string=getattr(client, "search_string", ""))
-                client.write_message(json.dumps(msg))
-            except Exception:
-                session.ws_clients.discard(client)
+        broadcast_state(session)
 
         klass_count = len(extra_klasses)
         log.info("reload_expr session=%s project_root=%s klasses=%d",

buckaroo/server/session.py

@@ -26,6 +26,25 @@ def dataflow_stats_tier(stats_tier: str, stats_delivery: str) -> str:
     return "schema" if stats_delivery == "deferred" else stats_tier
 
 
+# What ``df_meta.stats.status`` says about a session's stats for its current
+# ``stats_gen``: ``complete`` (the stats of the tier the session is headed for
+# are in the snapshot), ``pending`` (a deferred session that has not produced
+# them yet), ``not_computed`` (the session is headed for the schema tier, so
+# none will arrive) and ``error`` (the run failed; sticky until the next
+# generation).
+STATS_STATUSES = ("complete", "pending", "not_computed", "error")
+
+
+def initial_stats_status(stats_tier: str, stats_delivery: str) -> tuple[str, Optional[str]]:
+    """The status, and its reason, of a session whose stats generation has just
+    started, from the policy pair alone."""
+    if stats_tier != "full":
+        return "not_computed", "host"
+    if stats_delivery == "deferred":
+        return "pending", None
+    return "complete", None
+
+
 @dataclass
 class SessionState:
     session_id: str
@@ -64,6 +83,16 @@ class SessionState:
     # tier the dataflow is built at follows from the pair (dataflow_stats_tier).
     stats_tier: str = DEFAULT_STATS_TIER
     stats_delivery: str = DEFAULT_STATS_DELIVERY
+    # The stats generation: a counter the server owns, bumped whenever the
+    # dataflow state the stats describe changes (/load, /load_expr, /load_compare,
+    # /reload_expr, and a buckaroo_state_change that touches a dataflow field).
+    # It rides on df_meta.stats and on every stats message, so a client can drop
+    # a reply for a state it has left. Independent of any client-owned request
+    # sequence (state_seq). ``stats_status`` and ``stats_reason`` live here, not
+    # in ``df_meta``, because the dataflow rebuilds df_meta wholesale.
+    stats_gen: int = 0
+    stats_status: str = "complete"
+    stats_reason: Optional[str] = None
     # Companion telemetry sink (#943): a fire-and-forget POST callable, built
     # once from the /load_expr payload's telemetry_url on the IOLoop (where
     # make_http_sink captures AsyncHTTPClient/IOLoop.current()). Stored here so
@@ -111,6 +140,33 @@ def release_xorq_state(self) -> None:
 this field is the runtime escape hatch."""
 
 
+def begin_stats_generation(session: "SessionState") -> None:
+    """Start a new stats generation: the session's dataflow state has changed,
+    so stats for the previous one no longer describe it. The status restarts
+    from the session's policy pair."""
+    session.stats_gen += 1
+    session.stats_status, session.stats_reason = initial_stats_status(
+        session.stats_tier, session.stats_delivery)
+
+
+def stats_meta(session: "SessionState") -> Optional[dict]:
+    """The ``df_meta.stats`` value for a session: ``{status, tier, gen}`` plus
+    ``reason`` when there is one. ``tier`` is the tier reached so far, which is
+    the one the session is headed for only once it is complete.
+
+    ``None`` for a session on the default policy (inline delivery, full tier,
+    complete), which sends the message it always has; a client reads a missing
+    ``stats`` as complete."""
+    if session.stats_status == "complete" and session.stats_delivery != "deferred":
+        return None
+    stats: dict = {"status": session.stats_status,
+        "tier": session.stats_tier if session.stats_status == "complete" else "schema",
+        "gen": session.stats_gen}
+    if session.stats_reason:
+        stats["reason"] = session.stats_reason
+    return stats
+
+
 def build_state_message(session: "SessionState", metadata: dict | None = None,
                          search_string: str = "", reply_seq: int | None = None) -> dict:
     """Build the full ``initial_state`` WebSocket payload from a session.
@@ -137,10 +193,16 @@ def build_state_message(session: "SessionState", metadata: dict | None = None,
     Returns:
         A dict ready to be JSON-serialised and sent to WebSocket clients.
     """
+    # The dataflow rebuilds df_meta wholesale, so the stats status is injected
+    # here, into a copy, rather than stored in it.
+    df_meta = session.df_meta
+    stats = stats_meta(session)
+    if stats is not None:
+        df_meta = {**df_meta, "stats": stats}
     msg: dict = {"type": "initial_state", "protocol_version": PROTOCOL_VERSION,
         "metadata": metadata if metadata is not None else session.metadata,
         "prompt": session.prompt, "df_display_args": session.df_display_args, "df_data_dict": session.df_data_dict,
-        "df_meta": session.df_meta, "mode": session.mode}
+        "df_meta": df_meta, "mode": session.mode}
     if reply_seq is not None:
         msg["reply_seq"] = reply_seq
     if session.mode == "buckaroo":