fix(stats): surface summary-stat cache write side via telemetry (#951) (#952)

The xorq snapshot cache did its I/O with no operator-visible signal. The
firstpull.summary_stats span (#943) carried only cache_status/hits/misses/
secs, and _log_cache_stats emitted the write side (snapshots/bytes/
write_errors) at log.info to a server log file no deployment reads — so a
cache that stopped writing, or a run with write_errors > 0 (#910), was
invisible without counting .parquet files by hand.

- xorq_buckaroo: add cache_snapshots/cache_bytes/cache_write_errors to the
  firstpull.summary_stats span attrs (already available from
  cache_run_stats()), so a write_errors > 0 run reaches a telemetry consumer.
- xorq_stat_pipeline: attach the per-run cache outcome to the stat.xorq.total
  telemetry span so every process_table run — not just the firstpull — routes
  its cache outcome through a bound sink, independent of log level. The
  log.info summary line stays for local perf/debug sessions.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: paddymul

buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py

@@ -225,11 +225,27 @@ def _write_snapshot(self, path, result_df):
             self._cache_stats["write_errors"] += 1
             log.warning("xorq stat snapshot write failed for %s: %s", path, e)
 
-    def _log_cache_stats(self):
-        """One summary line per run when a snapshot cache is in play (#910)."""
+    def _log_cache_stats(self, span=None):
+        """Surface the per-run snapshot-cache outcome (#910, #951).
+
+        Two channels, so the write side is never invisible:
+
+        * ``span`` — attach the hit/miss/snapshot/byte/write-error counts to the
+          run's ``stat.xorq.total`` telemetry span. A bound telemetry sink then
+          carries the cache outcome (including ``write_errors``) to the operator
+          without a debug log build — the ``log.info`` line below lands in a
+          server log file no deployment reads (#951).
+        * ``log`` — one summary line per run for a local perf/debug session."""
         if self.cache_storage is None:
             return
         s = self._cache_stats
+        if span is not None:
+            cs = self.cache_run_stats()
+            span.set_attr(
+                cache_status=cs["status"], cache_hits=cs["hits"],
+                cache_misses=cs["misses"], cache_secs=cs["secs"],
+                cache_snapshots=cs["snapshots"], cache_bytes=cs["bytes"],
+                cache_write_errors=cs["write_errors"])
         base_path = getattr(getattr(self.cache_storage, "storage", None), "base_path", "?")
         log.info(
             "xorq stat cache [%s]: %d hit(s), %d miss(es), %d snapshot(s) "
@@ -318,12 +334,12 @@ def process_table(self, table, skip_columns=None) -> Tuple[SDType, List[StatErro
         self._perf = (perf_log.PerfRecorder()
                       if perf_log.enabled() and not self._suppress_perf_summary else None)
         _t0 = time.perf_counter()
-        with self._span("stat.xorq.total"):
+        with self._span("stat.xorq.total") as span:
             try:
                 return self._process_table_impl(table, skip_columns=skip_columns)
             finally:
                 self._cache_stats["secs"] = round(time.perf_counter() - _t0, 4)
-                self._log_cache_stats()
+                self._log_cache_stats(span)
                 if self._perf is not None:
                     self._perf.label = (
                         f"xorq cols={len(table.columns)} "

buckaroo/xorq_buckaroo.py

@@ -173,16 +173,22 @@ def _get_summary_sd(self, processed_df):
                 processed_df, self.analysis_klasses, self.df_name,
                 debug=self.debug, cache_storage=cache_storage,
                 skip_columns=getattr(self, 'skip_stat_columns', None))
-            # Attach the summary-stat cache hit/miss/timing signal (#944) to the
-            # span so a telemetry consumer learns whether the stats were cached
-            # — the one signal only the server observes (#943). The pandas branch
-            # above already returned, so stats here is always an XorqDfStatsV2,
-            # which always exposes cache_run_stats(); call it unconditionally so
-            # a missing signal fails loudly rather than silently vanishing.
+            # Attach the summary-stat cache signal (#944) to the span so a
+            # telemetry consumer learns whether the stats were cached — the one
+            # signal only the server observes (#943). Carries the write side too
+            # (snapshots/bytes/write_errors, #951) so a cache that stops writing —
+            # or a run with write_errors > 0 — is visible to a telemetry consumer
+            # rather than only to a server log no deployment reads. The pandas
+            # branch above already returned, so stats here is always an
+            # XorqDfStatsV2, which always exposes cache_run_stats(); call it
+            # unconditionally so a missing signal fails loudly rather than
+            # silently vanishing.
             cs = stats.cache_run_stats()
             span.set_attr(
                 cache_status=cs.get("status"), cache_hits=cs.get("hits"),
-                cache_misses=cs.get("misses"), cache_secs=cs.get("secs"))
+                cache_misses=cs.get("misses"), cache_secs=cs.get("secs"),
+                cache_snapshots=cs.get("snapshots"), cache_bytes=cs.get("bytes"),
+                cache_write_errors=cs.get("write_errors"))
         sdf = stats.sdf
         if stats.errs:
             if self.debug:

tests/unit/server/test_load_expr.py

@@ -445,6 +445,12 @@ async def test_load_expr_telemetry_emits_session_correlated_spans(self):
             self.assertEqual(attrs["cache_status"], "miss")
             self.assertEqual(attrs["cache_hits"], 0)
             self.assertGreater(attrs["cache_misses"], 0)
+            # The write side rides the span too (#951): the cold miss writes
+            # snapshots with no write errors, so a cache that stops writing — or a
+            # run with write_errors > 0 — is visible to the telemetry consumer.
+            self.assertGreater(attrs["cache_snapshots"], 0)
+            self.assertGreater(attrs["cache_bytes"], 0)
+            self.assertEqual(attrs["cache_write_errors"], 0)
         finally:
             shutil.rmtree(builds_root, ignore_errors=True)
             shutil.rmtree(cache_root, ignore_errors=True)

tests/unit/test_xorq_stats_v2.py

@@ -831,6 +831,31 @@ def test_internal_spans_emit_to_telemetry_sink_with_perf_off(self):
             f"stat.xorq.total span must emit to the sink with perf logging off; "
             f"got {names}")
 
+    def test_cache_outcome_rides_total_span_to_telemetry(self, tmp_path):
+        """#951: the per-run snapshot-cache outcome — status, snapshots, bytes,
+        write_errors — rides the stat.xorq.total telemetry span, so a telemetry
+        consumer sees the write side without reading a server log.
+
+        Before the fix _log_cache_stats emitted only a log.info line the server
+        never surfaced, so a cache that stopped writing was invisible."""
+        records: list = []
+        saved = perf_log._ENABLED
+        perf_log._ENABLED = False
+        try:
+            pipeline = XorqStatPipeline(
+                XORQ_STATS_V2, unit_test=False, cache_storage=self._cache(tmp_path))
+            with perf_log.telemetry_context("sess-cache-write", records.append):
+                pipeline.process_table(self._filter_chain_table())
+        finally:
+            perf_log._ENABLED = saved
+        total = next(r for r in records if r["name"] == "stat.xorq.total")
+        attrs = total["attrs"]
+        # Cold run against a fresh cache dir → a pure miss that writes snapshots.
+        assert attrs["cache_status"] == "miss"
+        assert attrs["cache_snapshots"] > 0
+        assert attrs["cache_bytes"] > 0
+        assert attrs["cache_write_errors"] == 0
+
     def test_cached_stats_match_uncached(self, tmp_path):
         """The cold (compute + write) and warm (snapshot-read) cache paths
         produce the same stats as a plain uncached run — the cache is a perf