diff --git a/pyproject.toml b/pyproject.toml index ffb56d9..bcc35c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "central" -version = "0.14.8" +version = "0.14.9" requires-python = ">=3.12,<3.13" description = "Data hub spine — adapters, bus, archive." readme = "README.md" diff --git a/src/central/gui/routes.py b/src/central/gui/routes.py index 8dff9c5..7258fb4 100644 --- a/src/central/gui/routes.py +++ b/src/central/gui/routes.py @@ -2224,6 +2224,8 @@ async def consumers_list(request: Request) -> HTMLResponse: "num_ack_pending": ci.num_ack_pending, "num_redelivered": ci.num_redelivered, "num_waiting": ci.num_waiting, + "delivered": getattr(getattr(ci, "delivered", None), "consumer_seq", None), + "acked": getattr(getattr(ci, "ack_floor", None), "consumer_seq", None), "created": ci.created, "protected": ci.name.startswith("archive-"), }) diff --git a/src/central/gui/templates/consumers_list.html b/src/central/gui/templates/consumers_list.html index 11cfac8..07693a5 100644 --- a/src/central/gui/templates/consumers_list.html +++ b/src/central/gui/templates/consumers_list.html @@ -25,6 +25,8 @@ has no active subscriber — it is safe to delete if it is not a central-owned c Ack Pending Redelivered Waiting + Delivered + Confirmed Created Action @@ -37,6 +39,8 @@ has no active subscriber — it is safe to delete if it is not a central-owned c {{ c.num_ack_pending if c.num_ack_pending is not none else '—' }} {{ c.num_redelivered if c.num_redelivered is not none else '—' }} {{ c.num_waiting if c.num_waiting is not none else '—' }} + {{ c.delivered if c.delivered is not none else '—' }} + {{ c.acked if c.acked is not none else '—' }} {{ c.created.isoformat() if c.created else '—' }} {% if c.protected %} @@ -64,7 +68,10 @@ has no active subscriber — it is safe to delete if it is not a central-owned c

Legend: Pending = messages not yet delivered to this consumer; Ack Pending = delivered but not yet acknowledged; - Waiting = active pull requests from a live subscriber. + Waiting = active pull requests from a live subscriber; + Delivered = total messages this consumer has received from the stream (lifetime counter); + Confirmed = total messages this consumer has acknowledged (processed) — these are consumer-level + counters (the subscriber got/acked it) and are NOT confirmation that the message reached a mesh device. A consumer with high Pending and zero Waiting is abandoned — no subscriber is pulling from it and messages are piling up. Consumers marked central-owned (archive-*) are managed by central and cannot be deleted here. diff --git a/tests/test_consumers.py b/tests/test_consumers.py index 743dcb8..95aa39c 100644 --- a/tests/test_consumers.py +++ b/tests/test_consumers.py @@ -2,6 +2,7 @@ import os from datetime import datetime, timezone +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -13,13 +14,16 @@ os.environ.setdefault("CENTRAL_NATS_URL", "nats://localhost:4222") def _make_consumer_info(name: str, num_pending: int = 0, num_ack_pending: int = 0, - num_redelivered: int = 0, num_waiting: int = 0): + num_redelivered: int = 0, num_waiting: int = 0, + delivered_seq: int = 0, ack_floor_seq: int = 0): ci = MagicMock() ci.name = name ci.num_pending = num_pending ci.num_ack_pending = num_ack_pending ci.num_redelivered = num_redelivered ci.num_waiting = num_waiting + ci.delivered = SimpleNamespace(consumer_seq=delivered_seq) + ci.ack_floor = SimpleNamespace(consumer_seq=ack_floor_seq) ci.created = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc) return ci @@ -204,6 +208,8 @@ class TestConsumersListHtmlRender: "num_ack_pending": 0, "num_redelivered": 0, "num_waiting": 0, + "delivered": 5000, + "acked": 4000, "created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc), "protected": False, }, @@ -230,6 +236,8 @@ class TestConsumersListHtmlRender: "num_ack_pending": 0, "num_redelivered": 0, "num_waiting": 1, + "delivered": 100, + "acked": 95, "created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc), "protected": True, }, @@ -254,6 +262,8 @@ class TestConsumersListHtmlRender: "num_ack_pending": None, "num_redelivered": None, "num_waiting": None, + "delivered": None, + "acked": None, "created": None, "protected": False, }, @@ -267,6 +277,34 @@ class TestConsumersListHtmlRender: # The guarded fallback em dash is rendered instead assert "—" in html + def test_delivered_and_confirmed_columns_render(self): + streams = [ + { + "stream": "CENTRAL_WX", + "error": None, + "consumers": [ + { + "name": "meshai-wx", + "num_pending": 10, + "num_ack_pending": 2, + "num_redelivered": 0, + "num_waiting": 1, + "delivered": 7777, + "acked": 6543, + "created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc), + "protected": False, + }, + ], + }, + ] + html = self._render(streams) + # Delivered and Confirmed column headers are present + assert "Delivered" in html + assert "Confirmed" in html + # The actual counter values appear in the rendered HTML + assert "7777" in html + assert "6543" in html + class TestConsumersDeleteArchiveGuard: """POST /consumers/{stream}/archive-*/delete must be refused."""