Say how much work is waiting, not just how much is running

`RedisWorkQueue.stats` read XPENDING, which counts entries delivered to a
consumer and not yet acknowledged — work in progress. Entries sitting in the
stream undelivered were counted nowhere, so an engine hours behind reported
itself idle: on the house, `pending: 4` while the group's lag was 1554.

The group's own `lag` is the missing number. `backlog` now carries it on both
queues (`len(_items)` in memory), leads the health tile, and a sustained one
publishes `engine_degraded` from the timer thread — named with the flow most
of the waiting work belongs to, sampled from the undelivered tail, since that
is the actionable half. It is a summary problem rather than a /utils/health
503: a backlog should not restart the container.

Also drops the keyspace `scan_iter` `stats()` did per poll to count parked
items — it walked every state and idempotency key twice per ten seconds — for
a set the park/unpark path maintains.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BpfSinyCBfjuieikyfMPbf

backend/fluksio/flow/queue.py

@@ -165,8 +165,23 @@ class WorkQueue(ABC):
 
     @abstractmethod
     def stats(self) -> dict[str, Any]:
-        """In-flight, delayed and parked counts plus the oldest pending age,
-        for the health endpoint."""
+        """Backlog, in-flight, delayed and parked counts plus the oldest
+        pending age, for the health endpoint."""
+
+    def backlog(self) -> int:
+        """How much work is waiting to be claimed.
+
+        Distinct from ``pending``, which is what has been handed to a consumer
+        and not yet acknowledged — work in progress. An engine hours behind has
+        a small ``pending`` and a large backlog, which is why one is not the
+        other. Concrete rather than abstract so the degradation watcher can ask
+        any queue.
+        """
+        return 0
+
+    def backlog_flows(self, sample: int = 100) -> dict[str, int]:
+        """Which flows the waiting work belongs to, as far as can be sampled."""
+        return {}
 
     @abstractmethod
     def dead_letters(self, count: int = 50) -> list[dict[str, Any]]:
@@ -273,9 +288,14 @@ class MemoryWorkQueue(WorkQueue):
         with self._lock:
             self._parked.pop(flow, None)
 
+    def backlog(self) -> int:
+        with self._lock:
+            return len(self._items)
+
     def stats(self) -> dict[str, Any]:
         with self._lock:
             return {
+                "backlog": len(self._items),
                 "pending": self._in_flight,
                 "delayed": len(self._delayed),
                 "parked": sum(len(v) for v in self._parked.values()),
@@ -322,6 +342,10 @@ class RedisWorkQueue(WorkQueue):
         self._stream = f"{namespace}:__queue__"
         self._delayed_key = f"{namespace}:__delayed__"
         self._dead_key = f"{namespace}:__dead__"
+        # Which flows have something parked. Maintained rather than discovered:
+        # finding them with a keyspace scan walked every state and idempotency
+        # key in the database, twice per health poll.
+        self._parked_flows_key = f"{namespace}:__parked_flows__"
         self._ensure_group()
 
     def _ensure_group(self) -> None:
@@ -443,20 +467,76 @@ class RedisWorkQueue(WorkQueue):
         logger.error("Dead-lettered work item for '%s': %s", item.node, reason)
 
     def park(self, flow: str, item: WorkItem) -> None:
-        self._redis.rpush(self._parked_key(flow), json.dumps(item.to_fields()))
+        pipe = self._redis.pipeline()
+        pipe.rpush(self._parked_key(flow), json.dumps(item.to_fields()))
+        pipe.sadd(self._parked_flows_key, flow)
+        pipe.execute()
 
     def unpark(self, flow: str) -> list[WorkItem]:
         key = self._parked_key(flow)
         raw = cast(list[str], self._redis.lrange(key, 0, -1))
-        self._redis.delete(key)
+        pipe = self._redis.pipeline()
+        pipe.delete(key)
+        pipe.srem(self._parked_flows_key, flow)
+        pipe.execute()
         return [WorkItem.from_fields(json.loads(r), "") for r in raw]
 
     def unpark_one(self, flow: str) -> WorkItem | None:
-        raw = cast("str | None", self._redis.lpop(self._parked_key(flow)))
+        key = self._parked_key(flow)
+        pipe = self._redis.pipeline()
+        pipe.lpop(key)
+        pipe.llen(key)
+        raw, remaining = cast(tuple["str | None", int], pipe.execute())
+        if not remaining:
+            self._redis.srem(self._parked_flows_key, flow)
         return WorkItem.from_fields(json.loads(raw), "") if raw else None
 
     def clear_flow(self, flow: str) -> None:
-        self._redis.delete(self._parked_key(flow))
+        pipe = self._redis.pipeline()
+        pipe.delete(self._parked_key(flow))
+        pipe.srem(self._parked_flows_key, flow)
+        pipe.execute()
+
+    def _group_info(self) -> dict[str, Any]:
+        """This consumer group's row of ``XINFO GROUPS``, or an empty one."""
+        try:
+            groups = cast(list[dict[str, Any]], self._redis.xinfo_groups(self._stream))
+        except redis.ResponseError:
+            # No stream yet: nothing has ever been enqueued.
+            return {}
+        return next((g for g in groups if g.get("name") == GROUP), {})
+
+    def backlog(self) -> int:
+        """Entries in the stream this group has never been handed.
+
+        Redis calls it the group's ``lag``. It is nil rather than zero when the
+        stream has been trimmed under the group — entries that were dropped
+        before anyone read them — and unknown is reported as none waiting,
+        since the alternative is a health screen crying wolf after a trim.
+        """
+        return int(self._group_info().get("lag") or 0)
+
+    def backlog_flows(self, sample: int = 100) -> dict[str, int]:
+        """Which flows the waiting work belongs to, from a sample of the tail.
+
+        The group's lag is one number for the whole stream, and the actionable
+        half of "the engine is behind" is always *what* is producing the work.
+        Reads forward from the last entry the group was handed, which is where
+        the undelivered entries start.
+        """
+        after = self._group_info().get("last-delivered-id")
+        if not after:
+            return {}
+        entries = cast(
+            list[tuple[str, dict[str, str]]],
+            self._redis.xrange(self._stream, min=f"({after}", max="+", count=sample),
+        )
+        counts: dict[str, int] = {}
+        for _entry_id, fields in entries:
+            flow = fields.get("flow", "")
+            if flow:
+                counts[flow] = counts.get(flow, 0) + 1
+        return counts
 
     def stats(self) -> dict[str, Any]:
         pending = cast(dict[str, Any], self._redis.xpending(self._stream, GROUP))
@@ -471,11 +551,10 @@ class RedisWorkQueue(WorkQueue):
             )
             if records:
                 oldest = records[0]["time_since_delivered"] / 1000.0
-        parked = sum(
-            cast(int, self._redis.llen(key))
-            for key in self._redis.scan_iter(f"{self._ns}:__parked__:*")
-        )
+        flows = cast(set[str], self._redis.smembers(self._parked_flows_key))
+        parked = sum(cast(int, self._redis.llen(self._parked_key(f))) for f in flows)
         return {
+            "backlog": self.backlog(),
             "pending": count,
             "delayed": cast(int, self._redis.zcard(self._delayed_key)),
             "parked": parked,