diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index a0e3aa8..5653153 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -148,6 +148,8 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: queue = await run_in_threadpool(controller.queue_stats) if queue.get("oldest_pending_s", 0) > 120: problems.append("queue stalled") + if queue.get("behind"): + problems.append(f"engine behind: {queue.get('backlog', 0)} items waiting") if queue.get("error"): problems.append("work queue unreachable") diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index e7d25c0..e3a4b47 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -41,6 +41,12 @@ DELAYED_INTERVAL_S = 1.0 MAX_CASCADES = 4 # How long a reload waits for claimed work to finish before rebuilding anyway. DRAIN_TIMEOUT_S = 10.0 +# Work waiting in the stream, undelivered. A burst is normal — the pool claims +# only what it can start — so what marks an engine as falling behind is the +# backlog staying up across several checks rather than any one reading. +BACKLOG_INTERVAL_S = 5.0 +BACKLOG_DEGRADED = 50 +BACKLOG_STRIKES = 3 class ExecutionService: @@ -70,6 +76,10 @@ class ExecutionService: ) self._consumer: threading.Thread | None = None self._timers: threading.Thread | None = None + # Consecutive backlog readings over the threshold, and whether the last + # of them said so out loud. + self._backlog_strikes = 0 + self.behind = False # ------------------------------------------------------------------------- # Lifecycle @@ -157,6 +167,7 @@ class ExecutionService: """Promote delayed items, and take back what a dead engine dropped.""" last_reclaim = 0.0 last_touch = 0.0 + last_backlog = 0.0 while not self._stop.is_set(): self._stop.wait(DELAYED_INTERVAL_S) if self._stop.is_set(): @@ -167,6 +178,13 @@ class ExecutionService: logger.error("Could not promote delayed work: %s", exc) now = time.monotonic() + if now - last_backlog >= BACKLOG_INTERVAL_S: + last_backlog = now + try: + self._check_backlog() + except Exception as exc: + logger.error("Could not read the queue backlog: %s", exc) + if now - last_touch >= TOUCH_INTERVAL_S: last_touch = now with self._inflight_lock: @@ -190,6 +208,38 @@ class ExecutionService: except Exception as exc: logger.error("Could not reclaim stale work: %s", exc) + def _check_backlog(self) -> None: + """Say so when work has been waiting in the stream for a while. + + A flow enqueuing faster than the pool drains produces no event of its + own: the backlog simply grows, every timer and connector poll drifts + behind it, and nothing on the health screen moves. This is that event. + The flow named is the one most of the waiting work belongs to, which is + the half somebody can act on. + """ + backlog = self.queue.backlog() + if backlog < BACKLOG_DEGRADED: + self._backlog_strikes = 0 + self.behind = False + return + + self._backlog_strikes += 1 + if self._backlog_strikes < BACKLOG_STRIKES or self.behind: + return + + self.behind = True + flows = self.queue.backlog_flows() + worst = max(flows, key=lambda f: flows[f], default="") + logger.warning("engine behind: %d items waiting (%s)", backlog, worst or "?") + self._publish( + { + "type": "engine_degraded", + "reason": f"{backlog} items waiting in the queue", + "flow": worst, + "ts": time.time(), + } + ) + def _await_capacity(self) -> int: """How many cascades may be claimed now. Zero means the service stops. @@ -366,6 +416,7 @@ class ExecutionService: return {"error": str(exc), "consumer_alive": self.alive()} stats["consumer_alive"] = self.alive() stats["cascades_busy"] = self._inflight + stats["behind"] = self.behind return stats def _publish_unavailable(self, exc: Exception) -> None: diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index d37c734..39baa61 100644 --- a/backend/fluksio/flow/queue.py +++ b/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, diff --git a/backend/scripts/bench_engine.py b/backend/scripts/bench_engine.py new file mode 100644 index 0000000..768cfdf --- /dev/null +++ b/backend/scripts/bench_engine.py @@ -0,0 +1,426 @@ +#!/usr/bin/env python +"""Engine throughput bench: what a message costs, in work and in time. + +`bench_startup.py` times submitting a run; this times the live path — a value +arriving at a subscriber and reaching everything downstream of it. It builds a +pipeline and an execution service in this process, so there is no HTTP, no +container and no browser between the measurement and the engine. + +Three scenarios, each answering a different question: + +``throttle`` + What does a rate-limited port cost? A source publishing every 15s behind + seven relays, driven at the rate the house's inverter actually sends. + Reports cascades enqueued, node executions and bus events per injection — + the numbers, not the wall clock, are the point. + +``throughput`` + How many messages a second, and how late is the last node? A relay chain, + driven until the queue drains, reporting end-to-end latency percentiles. + +``fanout`` + One source into many consumers: the scheduler's own overhead per wave. + +Run against the memory backends by default; ``--redis`` points state and queue +at a real Redis, which is what the engine runs on and where the round trips +this is meant to expose actually happen. +""" + +from __future__ import annotations + +import argparse +import cProfile +import json +import logging +import pstats +import statistics +import sys +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +from fluksio.flow.events import EventBus +from fluksio.flow.executor import ExecutionService +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes import Node +from fluksio.flow.pipeline import Pipeline +from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue +from fluksio.flow.state import MemoryState, RedisState, StateBackend + +log = logging.getLogger("bench") + +FLOW = "bench" + + +# ----------------------------------------------------------------------------- +# Counting what the engine did +# ----------------------------------------------------------------------------- + + +@dataclass +class Counters: + """What one scenario cost, counted rather than timed.""" + + executions: int = 0 + events: int = 0 + by_event: dict[str, int] = field(default_factory=dict) + #: inject -> when the last node downstream of it finished, in ms. + latencies: list[float] = field(default_factory=list) + + def count(self, event: dict[str, Any]) -> None: + kind = str(event.get("type", "")) + self.events += 1 + self.by_event[kind] = self.by_event.get(kind, 0) + 1 + + +class CountingBus(EventBus): + """The real bus, with a tally beside it. + + Subclassed rather than mocked: the cost of publishing — the queue per + subscriber, the drop-oldest — is part of what is being measured. + """ + + def __init__(self, counters: Counters) -> None: + super().__init__() + self._counters = counters + + def publish(self, event: dict[str, Any]) -> None: + self._counters.count(event) + super().publish(event) + + +# ----------------------------------------------------------------------------- +# Building graphs +# ----------------------------------------------------------------------------- + + +def spec(name: str, interval: float = 0.0) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT, interval=interval) + + +def node( + node_id: str, + f: Callable[..., Any], + requires: list[MessageSpec] | None = None, + provides: list[MessageSpec] | None = None, +) -> Node: + n = Node(f=f, requires=requires or [], provides=provides or [], name=node_id) + n.assign_flow(FLOW, node_id) + return n + + +def relay( + node_id: str, src: str, dst: str, interval: float, counters: Counters +) -> Node: + """A node that republishes what it read — the shape the house is full of. + + ``src`` and ``dst`` are bare port names: the pipeline qualifies them with + the flow, and the node function only ever sees ports. + """ + + def body(**kwargs: Any) -> dict[str, Any]: + counters.executions += 1 + return {dst: kwargs[src]} + + return node( + node_id, + body, + requires=[spec(src)], + provides=[spec(dst, interval=interval)], + ) + + +# ----------------------------------------------------------------------------- +# Harness +# ----------------------------------------------------------------------------- + + +class Engine: + """A pipeline and the service that drains its queue, as the app wires them.""" + + def __init__(self, nodes: list[Node], redis_host: str, counters: Counters) -> None: + self.counters = counters + self.bus = CountingBus(counters) + self.state: StateBackend + self.queue: WorkQueue + if redis_host: + # A namespace of its own, so a bench never touches a live engine's + # state or its stream. + self.state = RedisState(host=redis_host, namespace="bench") + self.queue = RedisWorkQueue(host=redis_host, namespace="benchq") + self.state.clear() + else: + self.state = MemoryState() + self.queue = MemoryWorkQueue() + self.service = ExecutionService(self.queue, max_workers=4, events=self.bus) + self.pipeline = Pipeline( + nodes=nodes, + state=self.state, + events=self.bus, + work_queue=self.queue, + node_pool=self.service.node_pool, + ) + self.service.bind(self.pipeline) + + def start(self) -> None: + self.service.start() + + def stop(self) -> None: + self.service.stop() + + def drain(self, timeout: float = 30.0) -> bool: + """Wait for the queue to go quiet. False means it never did. + + Delayed items are not waited for: a rate-limit flush is due a whole + window from now, and the scenario is over long before that. + """ + deadline = time.monotonic() + timeout + quiet_since = 0.0 + while time.monotonic() < deadline: + stats = self.service.stats() + busy = stats.get("cascades_busy", 0) or stats.get("backlog", 0) + if not busy: + # Two consecutive quiet reads: a cascade between claim and ack + # shows as neither. + if quiet_since and time.monotonic() - quiet_since > 0.15: + return True + quiet_since = quiet_since or time.monotonic() + else: + quiet_since = 0.0 + time.sleep(0.02) + return False + + +def percentile(values: list[float], p: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + idx = min(len(ordered) - 1, int(round(p / 100.0 * (len(ordered) - 1)))) + return ordered[idx] + + +# ----------------------------------------------------------------------------- +# Scenarios +# ----------------------------------------------------------------------------- + + +def scenario_throttle(args: argparse.Namespace) -> dict[str, Any]: + """A rate-limited source behind relays, driven far faster than its window. + + The house's shape: an inverter sending ~200 readings a minute into seven + relay nodes, each told to publish every 15s. What should cost seven + publishes a minute is what this counts. + """ + counters = Counters() + source = node( + "source", + lambda params: None, + provides=[spec("raw", interval=args.window)], + ) + nodes = [source] + for i in range(args.relays): + nodes.append(relay(f"relay{i}", "raw", f"out{i}", args.window, counters)) + + engine = Engine(nodes, args.redis, counters) + engine.start() + try: + started = time.perf_counter() + for i in range(args.messages): + source.inject({"raw": float(i)}) + injected = time.perf_counter() - started + drained = engine.drain() + finally: + engine.stop() + + return { + "scenario": "throttle", + "messages": args.messages, + "relays": args.relays, + "window_s": args.window, + "drained": drained, + "inject_s": round(injected, 3), + # The three numbers the bug is about. + "cascades_started": counters.by_event.get("cascade_started", 0), + "node_executions": counters.executions, + "bus_events": counters.events, + "published": counters.by_event.get("message_value", 0), + "per_message": { + "cascades": round( + counters.by_event.get("cascade_started", 0) / args.messages, 3 + ), + "executions": round(counters.executions / args.messages, 3), + "events": round(counters.events / args.messages, 3), + }, + } + + +def scenario_throughput(args: argparse.Namespace) -> dict[str, Any]: + """A relay chain, driven flat out: messages a second, and how late the tail is.""" + counters = Counters() + starts: dict[float, float] = {} + lock = threading.Lock() + + source = node("source", lambda params: None, provides=[spec("v0")]) + nodes = [source] + for i in range(args.depth): + nodes.append(relay(f"relay{i}", f"v{i}", f"v{i + 1}", 0.0, counters)) + + # The last link records how long the value took to walk the whole chain. + last_port = f"v{args.depth}" + + def sink(**kwargs: Any) -> None: + counters.executions += 1 + value = kwargs[last_port] + with lock: + sent = starts.get(value) + if sent is not None: + counters.latencies.append((time.perf_counter() - sent) * 1000) + + nodes.append(node("sink", sink, requires=[spec(last_port)])) + + engine = Engine(nodes, args.redis, counters) + engine.start() + try: + started = time.perf_counter() + for i in range(args.messages): + value = float(i) + with lock: + starts[value] = time.perf_counter() + source.inject({"v0": value}) + drained = engine.drain() + elapsed = time.perf_counter() - started + finally: + engine.stop() + + delivered = len(counters.latencies) + return { + "scenario": "throughput", + "messages": args.messages, + "depth": args.depth, + "drained": drained, + "elapsed_s": round(elapsed, 3), + "msgs_per_s": round(args.messages / elapsed, 1) if elapsed else 0.0, + "delivered": delivered, + "latency_ms": { + "p50": round(percentile(counters.latencies, 50), 2), + "p95": round(percentile(counters.latencies, 95), 2), + "max": round(max(counters.latencies), 2) if counters.latencies else 0.0, + "mean": round(statistics.fmean(counters.latencies), 2) + if delivered + else 0.0, + }, + "node_executions": counters.executions, + "bus_events": counters.events, + } + + +def scenario_fanout(args: argparse.Namespace) -> dict[str, Any]: + """One source into many consumers: what the wave scheduler costs per node.""" + counters = Counters() + source = node("source", lambda params: None, provides=[spec("raw")]) + nodes = [source] + for i in range(args.width): + nodes.append(relay(f"leaf{i}", "raw", f"out{i}", 0.0, counters)) + + engine = Engine(nodes, args.redis, counters) + engine.start() + try: + started = time.perf_counter() + for i in range(args.messages): + source.inject({"raw": float(i)}) + drained = engine.drain() + elapsed = time.perf_counter() - started + finally: + engine.stop() + + return { + "scenario": "fanout", + "messages": args.messages, + "width": args.width, + "drained": drained, + "elapsed_s": round(elapsed, 3), + "msgs_per_s": round(args.messages / elapsed, 1) if elapsed else 0.0, + "node_executions": counters.executions, + "bus_events": counters.events, + } + + +SCENARIOS: dict[str, Callable[[argparse.Namespace], dict[str, Any]]] = { + "throttle": scenario_throttle, + "throughput": scenario_throughput, + "fanout": scenario_fanout, +} + + +# ----------------------------------------------------------------------------- +# Entry point +# ----------------------------------------------------------------------------- + + +def report(result: dict[str, Any]) -> None: + name = result.pop("scenario") + log.info("\n=== %s ===", name) + for key, value in result.items(): + if isinstance(value, dict): + inner = " ".join(f"{k}={v}" for k, v in value.items()) + log.info(" %-18s %s", key, inner) + else: + log.info(" %-18s %s", key, value) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--scenario", + choices=[*SCENARIOS, "all"], + default="all", + help="Which scenario to run (default: all).", + ) + parser.add_argument("--messages", type=int, default=500) + parser.add_argument("--relays", type=int, default=7, help="throttle: relay count") + parser.add_argument("--window", type=float, default=15.0, help="throttle: interval") + parser.add_argument("--depth", type=int, default=5, help="throughput: chain length") + parser.add_argument("--width", type=int, default=20, help="fanout: consumer count") + parser.add_argument( + "--redis", + nargs="?", + const="localhost", + default="", + metavar="HOST", + help="Run against a real Redis (namespaces bench/benchq) rather than memory.", + ) + parser.add_argument("--profile", action="store_true", help="cProfile, top 25.") + parser.add_argument("--json", action="store_true", help="Machine-readable output.") + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout) + + names = list(SCENARIOS) if args.scenario == "all" else [args.scenario] + + profiler = cProfile.Profile() if args.profile else None + if profiler: + profiler.enable() + + results = [SCENARIOS[name](args) for name in names] + + if profiler: + profiler.disable() + + if args.json: + log.info(json.dumps(results, indent=2)) + else: + for result in results: + report(dict(result)) + + if profiler: + log.info("\n=== profile (cumulative, top 25) ===") + pstats.Stats(profiler, stream=sys.stdout).sort_stats("cumulative").print_stats( + 25 + ) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 7340715..6d7620f 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -78,6 +78,49 @@ def test_claimed_work_counts_as_in_flight_until_it_is_acknowledged(): assert queue.stats()["pending"] == 0 +def test_work_waiting_to_be_claimed_is_the_backlog(): + """`pending` is what is running; an engine hours behind reports it as idle.""" + queue = MemoryWorkQueue() + queue.add(WorkItem(kind="cascade", node="f.n", flow="f")) + + assert queue.stats()["backlog"] == 1 + assert queue.stats()["pending"] == 0 + + (item,) = queue.claim(1, 10) + assert queue.stats()["backlog"] == 0 + assert queue.stats()["pending"] == 1 + + queue.ack(item) + assert queue.stats()["backlog"] == 0 + + +def test_a_sustained_backlog_says_the_engine_is_behind(): + """A flow enqueuing faster than the pool drains produced no signal at all.""" + events: list[dict] = [] + queue = MemoryWorkQueue() + service = ExecutionService(queue) + service._publish = events.append # type: ignore[method-assign] + for _ in range(executor.BACKLOG_DEGRADED): + queue.add(WorkItem(kind="cascade", node="f.n", flow="f")) + + for _ in range(executor.BACKLOG_STRIKES - 1): + service._check_backlog() + assert events == [] + + service._check_backlog() + assert [e["type"] for e in events] == ["engine_degraded"] + assert service.stats()["behind"] is True + + # Said once, not once every five seconds for as long as it lasts. + service._check_backlog() + assert len(events) == 1 + + # And a drained queue clears it, so the next backlog is announced again. + queue.claim(executor.BACKLOG_DEGRADED, 10) + service._check_backlog() + assert service.stats()["behind"] is False + + def _pipeline_with_a_consumer() -> tuple[Pipeline, Node, MemoryState, list]: """A source whose message a consumer records.""" seen: list[float] = [] @@ -383,6 +426,8 @@ def test_no_more_is_claimed_than_the_pool_can_run(): assert stats["cascades_busy"] <= executor.MAX_CASCADES # And the journal entries of what is only waiting are still free. assert stats["pending"] <= executor.MAX_CASCADES + # Waiting is not idle: the rest of the forty is the backlog. + assert stats["backlog"] >= 30 finally: release.set() service.stop() diff --git a/frontend/src/components/Health/HealthOverview.tsx b/frontend/src/components/Health/HealthOverview.tsx index 552cb36..a206434 100644 --- a/frontend/src/components/Health/HealthOverview.tsx +++ b/frontend/src/components/Health/HealthOverview.tsx @@ -144,10 +144,13 @@ export function HealthOverview({ : "nothing recorded" } /> + {/* Backlog leads: what is waiting is what says the engine is + behind. `pending` is work already running, which reads as idle + on an engine hours behind. */}