Bound the rate-aware batcher backlog

src/ess/livedata/core/message_batcher.py

@@ -32,6 +32,30 @@ class MessageBatch:
     messages: list[Message[Any]]
 
 
+@dataclass(frozen=True, slots=True)
+class BatcherMetrics:
+    """Backlog observability for the periodic service metrics line.
+
+    A batcher that retains a backlog is the one place where lag between live
+    data and what the service delivers is visible: the Kafka consumer reports
+    no lag for data it has already handed over, and per-stream ingest lag is
+    measured before batching.  ``max_backlog_s`` is therefore a leading
+    indicator -- it rises during any stall, whether or not shedding follows.
+
+    All fields cover the interval since the last drain, like the per-interval
+    counters in the periodic ``consumer_metrics`` line.
+    """
+
+    max_backlog_s: float
+    """Deepest backlog observed, in data time, since the last drain."""
+
+    dropped_messages: int
+    """Messages shed to keep the backlog bounded, since the last drain."""
+
+    dropped_bytes: int
+    """Payload bytes shed to keep the backlog bounded, since the last drain."""
+
+
 # Largest jump between consecutive timestamps that still reads as continuous
 # traffic, in multiples of the batch length.  Data-derived timestamps are the
 # batchers' only clock, so window placement must not follow one outlier
@@ -114,6 +138,18 @@ def batch_length_s(self) -> float:
         """Current effective batch length in seconds."""
         return 1.0
 
+    def drain_metrics(self) -> BatcherMetrics:
+        """Return backlog metrics for the interval since the last drain.
+
+        The default is an empty snapshot, accurate for batchers that hold no
+        more than the active window and so cannot fall behind live data.
+        ``SimpleMessageBatcher`` is the known exception: its future-message
+        backlog is unbounded, yet it reports zeros here -- it is a fallback
+        pending replacement by the rate-aware batcher and deliberately does
+        not implement the backlog bound.
+        """
+        return BatcherMetrics(max_backlog_s=0.0, dropped_messages=0, dropped_bytes=0)
+
 
 class NaiveMessageBatcher(MessageBatcher):
     """
@@ -362,6 +398,9 @@ def __init__(
     def batch(self, messages: list[Message[Any]]) -> MessageBatch | None:
         return self._inner.batch(messages)
 
+    def drain_metrics(self) -> BatcherMetrics:
+        return self._inner.drain_metrics()
+
     def report_batch(
         self,
         message_count: int | None,

src/ess/livedata/core/orchestrating_processor.py

@@ -378,6 +378,7 @@ def _maybe_log_metrics(self) -> None:
                 )
                 self._log_stream_lag(self._stream_stats_provider.drain_lag())
 
+            batcher_metrics = self._message_batcher.drain_metrics()
             logger.info(
                 'processor_metrics',
                 batches=self._batches_processed,
@@ -386,6 +387,13 @@ def _maybe_log_metrics(self) -> None:
                 errors=self._errors_since_last_metrics,
                 interval_seconds=window_seconds,
                 stream_stats=self._pending_stream_stats,
+                # How far behind live data the batcher fell, and what it shed
+                # to stay bounded.  A rising backlog is the earliest warning
+                # that the service is not keeping up; drops mean data for the
+                # shed interval never reached any workflow.
+                max_backlog_s=round(batcher_metrics.max_backlog_s, 3),
+                dropped_messages=batcher_metrics.dropped_messages,
+                dropped_bytes=batcher_metrics.dropped_bytes,
             )
             self._batches_processed = 0
             self._empty_batches = 0