Switch ingress/egress telemetry to pub/sub

AGENTS.md

@@ -37,6 +37,7 @@ Run a single test by node ID, e.g. `uv run pytest tests/test_command.py::TestCom
   ```bash
   curl -sSL https://raw.githubusercontent.com/codingjoe/naming-things/refs/heads/main/README.md
   ```
+- Use Google-style docstrings for mkdocs. NEVER use RST directives (`:param:`, `:returns:`, `:meth:`).
 - NEVER format code, tests, or docs manually. MUST USE `prek run --all-files`.
 
 ## Coverage

tests/backends/test_base.py

@@ -81,10 +81,10 @@ def test_peek__raise_not_implemented_error(self) -> None:
                 )
             )
 
-    def test_telemetry__raise_not_implemented_error(self) -> None:
-        """Raise NotImplementedError for backend telemetry API."""
+    async def test_queue_stats__raise_not_implemented_error(self) -> None:
+        """Raise NotImplementedError for backend queue_stats API."""
         with pytest.raises(NotImplementedError):
-            BackendDouble(alias="default", params={}).telemetry()
+            await BackendDouble(alias="default", params={}).queue_stats()
 
     def test_dequeue__raise_not_implemented_error(self) -> None:
         """Raise NotImplementedError for backend dequeue API."""

tests/conftest.py

@@ -11,7 +11,9 @@ def _flush_keys(backend) -> None:
 
 @pytest.fixture(autouse=True)
 def flush_default_backend():
-    """Flush all threadmill keys before and after each test."""
+    """Flush all threadmill keys and reset async client before and after each test."""
     _flush_keys(default_task_backend)
+    default_task_backend._async_client = None
     yield
     _flush_keys(default_task_backend)
+    default_task_backend._async_client = None

tests/test_inspector.py

@@ -37,16 +37,16 @@
 
 
 class FailingBackend(ThreadmillTaskBackend):
-    """Backend double whose peek and telemetry raise for error-path tests."""
+    """Backend double whose peek and queue_stats raise for error-path tests."""
 
     def enqueue(self, task, args, kwargs):
         raise NotImplementedError
 
     def peek(self, *args, **kwargs):
         raise RuntimeError("peek failed")
 
-    def telemetry(self, *, interval=None):
-        raise RuntimeError("telemetry failed")
+    async def queue_stats(self, *, interval=None):
+        raise RuntimeError("queue_stats failed")
 
 
 class ErroringBackend(RedisTaskBackend):
@@ -245,20 +245,20 @@ async def test_list_view_selected_sets_queue(self):
             task_list = app.query_one("#task-list", TaskList)
             assert task_list.queue_name == item.queue_name
 
-    async def test_telemetry_refresh_updates_and_prunes_queues(self):
+    async def test_queue_stats_refresh_updates_and_prunes_queues(self):
         """Telemetry refresh updates existing queue labels and drops stale queues."""
         app = InspectorApp(backend=default_task_backend, auto_refresh=False)
         async with app.run_test() as pilot:
             await pilot.pause()
-            app._refresh_telemetry()
+            app.telemetry = await app.backend.queue_stats()
             await pilot.pause()
             queue_list = app.query_one("#queue-list", QueueList)
             assert set(queue_list._items) == set(default_task_backend.queues)
             app.telemetry = BackendTelemetry(queues={"default": _stats()})
             await pilot.pause()
             assert list(queue_list._items) == ["default"]
 
-    async def test_telemetry_counts_are_scoped_to_selected_queue(self):
+    async def test_queue_stats_counts_are_scoped_to_selected_queue(self):
         """Tab counts reflect the selected queue, not backend-wide totals."""
         default_task_backend.enqueue(echo, args=[1])
         stats = _stats(ready=1)
@@ -457,17 +457,17 @@ async def test_successful_tab_lists_finished_task(self):
             assert table.row_count >= 1
 
     async def test_auto_refresh_timer_armed(self):
-        """With auto-refresh on, a telemetry timer is armed on mount."""
+        """With auto-refresh on, the sparkline timer is armed on mount."""
         app = InspectorApp(backend=default_task_backend)
         async with app.run_test() as pilot:
             await pilot.pause()
             await pilot.pause()
-            assert app._telemetry_timer is not None
-            assert app._telemetry_timer.name == "telemetry-refresh"
+            assert app._sparkline_timer is not None
+            assert app._sparkline_timer.name == "sparkline-refresh"
         app = InspectorApp(backend=default_task_backend, auto_refresh=False)
         async with app.run_test() as pilot:
             await pilot.pause()
-            assert app._telemetry_timer is None
+            assert app._sparkline_timer is None
 
     async def test_initial_focus_on_queue_list(self):
         """The app opens with focus on the queue list, not the backend selector."""
@@ -478,7 +478,7 @@ async def test_initial_focus_on_queue_list(self):
             assert app.focused is not None
             assert app.focused.id == "queue-list"
 
-    async def test_telemetry_refresh_does_not_re_peek_task_list(self):
+    async def test_queue_stats_refresh_does_not_re_peek_task_list(self):
         """A telemetry refresh updates counts but leaves task rows stale until manual refresh."""
         default_task_backend.enqueue(echo, args=[1])
         default_task_backend.enqueue(echo, args=[2])
@@ -495,7 +495,7 @@ async def test_telemetry_refresh_does_not_re_peek_task_list(self):
             default_task_backend.acquire(
                 timeout=datetime.timedelta(seconds=1), worker="stale-test"
             )
-            app.telemetry = app.backend.telemetry()
+            app.telemetry = await app.backend.queue_stats()
             await pilot.pause()
             await pilot.pause()
             assert table.row_count == before

threadmill/backends/base.py

@@ -1,6 +1,7 @@
 import collections.abc
 import dataclasses
 import datetime
+import enum
 import json
 import threading
 import typing
@@ -95,6 +96,21 @@ class BackendTelemetry:
     queues: dict[str, QueueStats]
 
 
+class TelemetryDirection(enum.Enum):
+    """Direction of a telemetry event: a task entering or leaving a queue."""
+
+    INGRESS = "ingress"
+    EGRESS = "egress"
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class TelemetryEvent:
+    """A single ingress/egress event published by a worker via pub/sub."""
+
+    direction: TelemetryDirection
+    queue_name: str
+
+
 class Broker(threading.Thread):
     """Backend maintenance thread launched by the task executor."""
 
@@ -242,14 +258,16 @@ def peek(
         """
         raise NotImplementedError
 
-    def telemetry(
+    async def queue_stats(
         self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60)
     ) -> BackendTelemetry:
-        """Return a snapshot of stats for all configured queues.
+        """Return per-queue task counts for all configured queues."""
+        raise NotImplementedError
 
-        Args:
-            interval: The time window for rolling rates.
-        """
+    async def worker_telemetry(
+        self,
+    ) -> collections.abc.AsyncIterator[TelemetryEvent]:
+        """Yield ingress/egress telemetry events from the backend's pub/sub stream."""
         raise NotImplementedError
 
     def dequeue(self, task_result: TaskResult) -> None:

threadmill/backends/lua/acknowledge.lua

@@ -3,7 +3,8 @@
 -- per-status results history set. Evicts results whose finish score falls
 -- outside the retention window (result_ttl). The per-status history sets are
 -- the time series the inspector counts over for telemetry, so no separate
--- egress window or status counters are needed here.
+-- egress window or status counters are needed here. A pub/sub event is
+-- published so the inspector can track live egress throughput.
 --
 -- KEYS[1]  -- running set (ZSET)
 -- KEYS[2]  -- result key (STRING, stores serialized TaskResult)
@@ -15,6 +16,8 @@
 -- ARGV[3]  -- result TTL in seconds
 -- ARGV[4]  -- finish timestamp in milliseconds (score for the history set)
 -- ARGV[5]  -- status (SUCCESSFUL or FAILED)
+-- ARGV[6]  -- telemetry pub/sub channel
+-- ARGV[7]  -- queue name
 -- Returns: 1 on success, 0 if task was not in the running set
 
 local removed = redis.call('ZREM', KEYS[1], ARGV[1])
@@ -31,4 +34,5 @@ if ARGV[5] ~= 'SUCCESSFUL' then
 end
 redis.call('ZADD', results_key, finish, ARGV[1])
 redis.call('ZREMRANGEBYSCORE', results_key, 0, cutoff)
+redis.call('PUBLISH', ARGV[6], 'egress:' .. ARGV[7])
 return 1