feat: readings kept through a storage stall; an agent's backlog arrives in order (#5995)

On 2026-10-17 the NFS under Postgres stalled for ~15 minutes. Two things
went wrong for the readings that would have explained it:

- The SNMP poll stored before it polled again, so while the database hung
  it stopped polling. It now collects on schedule and hands each round to a
  Holding (steward/core/holding.py): one writer stores what is held, oldest
  first, each reading at the time it was collected. Bounded at two hours of
  polls, with a 30-minute deadline on a store; the device list read falls
  back to the last one read after 10 s.
- "Mount not answering" (10 checks in a row) went pending on Docker02 and
  Docker03 and never fired: the agent sent each post's backlog with its
  newest, healthy sample, and that sample ended every post's run. Agent
  1.22.0 queues the new sample behind the backlog, so Steward receives a
  host's samples in the order they were taken.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/plugins/host-agent-design.md

@@ -327,7 +327,7 @@ The agent's self-reported hostname in the payload is advisory. The real identity
 ### Failure behavior
 
 - POST fails → push sample to ring buffer (max 20), exponential backoff `30s → 60s → 120s → 300s cap`, resets on first success.
-- On next success → flush all buffered samples in a single batched POST.
+- On next success → flush the buffer oldest first, five samples a POST (`MAX_SAMPLES_PER_POST`). The new sample queues behind the backlog, so Steward always receives a host's samples in the order they were taken (1.22.0, #5995): a rule counting checks in a row then sees a stall's run of breaching checks unbroken. Before, each post carried the newest sample too, and after the 2026-10-17 NFS stall its healthy reading ended every post's run, so "Mount not answering" (10 in a row) never fired.
 - Ring buffer full → drop oldest, keep newest. Logged at DEBUG (expected during extended outages).
 - Malformed payload rejection (400) → drop the sample (don't re-buffer, it'll just be rejected again). Indicates agent/server version skew.
 - Token rejection (401) → log ERROR with remediation hint, continue retry loop at backoff cadence. Admin fixes via UI + conf file edit; no restart needed.
@@ -461,7 +461,7 @@ Content-Type: application/json
 
 ### Design points
 
-- **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs `[current, ...buffered]` in a single request. One-path server code.
+- **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs the oldest buffered samples, with the current one only once the backlog is through. One-path server code.
 - **`ts` is agent-reported, ISO-8601 UTC.** Server trusts it for `PluginMetric.recorded_at`. Host clocks drift but NTP is reliable enough for our purposes.
 - **If `|ts - now| > 5 minutes`**, server logs a WARN but still accepts. Don't reject, don't lose data over clock skew.
 - **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts.

plugins/host_agent/agent.py

@@ -22,7 +22,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.21.0"
+AGENT_VERSION = "1.22.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -1956,10 +1956,16 @@ def _drop_logs(sample: dict) -> dict:
     return sample
 
 
-def split_backlog(buffered: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple:
-    """(the buffered samples to send with this one, the ones held for later):
-    the oldest limit-1, leaving room for the new sample."""
-    return buffered[:limit - 1], buffered[limit - 1:]
+def split_backlog(queued: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple:
+    """(the samples to send in this post, the ones held for later): the oldest
+    `limit` of the backlog with the new sample queued behind it.
+
+    The new sample waits its turn. Sent ahead of the backlog it reached Steward
+    before older samples did, and a rule counting checks in a row saw each
+    post's run of stalled checks ended by the newest, healthy one: after the
+    2026-10-17 NFS stall "Mount not answering" (10 in a row) never fired
+    (#5995)."""
+    return queued[:limit], queued[limit:]
 
 
 def next_backoff(current: int) -> int:
@@ -2074,10 +2080,10 @@ def main_loop(conf_path: str) -> int:
                               docker_logs_enabled=docker_logs_enabled,
                               docker_log_exclude=docker_log_exclude)
         buffered = buffer.drain()
-        send, held = split_backlog(buffered)
+        send, held = split_backlog(buffered + [sample])
         maint_events, maint_keys = maintenance_results(maint_seen)
         payload = build_payload(
-            samples=send + [sample],
+            samples=send,
             hostname=hostname,
             metadata=metadata,
             events=events + ([lost_contact_event(lost_since)] if lost_since else [])
@@ -2098,7 +2104,7 @@ def main_loop(conf_path: str) -> int:
                 rate_state["updates_due"] = True
             for s in held:
                 buffer.push(s)
-            if send:
+            if buffered:
                 _log("INFO", f"flushed {len(send)} buffered samples"
                              + (f", {len(held)} still held" if held else ""))
             backoff = 0

plugins/snmp/scheduler.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 from typing import TYPE_CHECKING
 
 from sqlalchemy import select
@@ -47,15 +49,75 @@ def make_poll_task(app: "Quart") -> ScheduledTask:
     )
 
 
-async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
-    """Poll every device at once, then write each one's outcome to its row
-    and its answers to the metric tables. A device whose vendor tables are
-    due is walked first, so a new disk is read in the same poll. `poll` and
-    `walk` stand in for poller.poll_device and poller.walk_tables in tests."""
-    global RATES
-    from steward.core.alerts import MetricBatch
+# Polls collected but not yet stored, written one store at a time (#5995): a
+# stall of the storage under Postgres delays them instead of losing them. Two
+# hours at the default one-minute poll.
+HELD_POLLS = 120
+# How long the device list may take to read before the last one read stands
+# in, so a stalled database doesn't stop the polling it is about to explain.
+DEVICES_READ_S = 10.0
+# How long a poll waits for its own store before collecting on.
+STORE_WAIT_S = 20.0
 
+# Each device as last read: plain copies, so a poll can go on without the
+# database, and a walk's rows found meanwhile can be kept on them.
+DEVICES: list | None = None
+
+
+@dataclass
+class Polled:
+    """One device's poll, collected and waiting to be stored."""
+    at: datetime
+    device_id: str
+    name: str
+    failed: str | None = None            # the poll raised: why, in words
+    error: str | None = None
+    scan: object = None                  # a TableScan when the tables were walked
+    answered: bool = False
+    missing: list = field(default_factory=list)
+    texts: dict = field(default_factory=dict)
+    values: list = field(default_factory=list)    # (metric, value)
+
+
+def _snapshot(device) -> SimpleNamespace:
+    from sqlalchemy import inspect as sa_inspect
+    return SimpleNamespace(**{a.key: getattr(device, a.key)
+                              for a in sa_inspect(type(device)).column_attrs})
+
+
+async def _devices(app: "Quart") -> list:
+    """The devices to poll: read now, or as last read when the database
+    doesn't answer within DEVICES_READ_S."""
+    global DEVICES
     from .models import SnmpDevice
+
+    async def read() -> list:
+        async with app.db_sessionmaker() as session:
+            return [_snapshot(d) for d in (await session.execute(select(SnmpDevice))).scalars()]
+
+    try:
+        DEVICES = await asyncio.wait_for(read(), DEVICES_READ_S)
+    except Exception as exc:
+        if DEVICES is None:
+            logger.warning("SNMP devices not read (%s); nothing to poll yet", exc or type(exc).__name__)
+            return []
+        logger.warning("SNMP devices not read within %.0f s (%s); polling the %d last read",
+                       DEVICES_READ_S, exc or type(exc).__name__, len(DEVICES))
+    return DEVICES
+
+
+async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
+    """Poll every device at once, hold what each answered, and store it: the
+    device's outcome on its row, its answers in the metric tables, each at
+    the time it was collected. A device whose vendor tables are due is walked
+    first, so a new disk is read in the same poll. When the database doesn't
+    take the store within STORE_WAIT_S the poll returns and the next one
+    collects on; what is held is stored, oldest first, once it does (#5995).
+    `poll` and `walk` stand in for poller.poll_device and poller.walk_tables
+    in tests."""
+    global RATES
+    from steward.core.holding import holding
+
     from .poller import poll_device, walk_tables
     from .readings import state_of, tables_for
     from .service import RateKeeper, credentials, readings_for
@@ -64,12 +126,10 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
     walk = walk or walk_tables
     if RATES is None:
         RATES = RateKeeper()
+    held = holding(app, "SNMP poll", store=lambda rounds: _store(app, rounds), limit=HELD_POLLS)
 
-    async with app.db_sessionmaker() as session:
-        devices = (await session.execute(select(SnmpDevice))).scalars().all()
-    if not devices:
-        return
-
+    devices = await _devices(app)
+    if devices:
         started = datetime.now(timezone.utc)
         scans, asked = {}, {}
 
@@ -84,26 +144,20 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
         results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True)
 
         now = datetime.now(timezone.utc)
-    async with app.db_sessionmaker() as session:
-        async with session.begin():
-            batch = MetricBatch()
+        polls = []
         for device, result in zip(devices, results):
-                row = await session.get(SnmpDevice, device.id)
-                if row is None:          # deleted while it was being polled
-                    continue
-                row.last_polled_at = now
             if isinstance(result, BaseException):
                 logger.warning("SNMP poll of %s failed: %s", device.name, result)
-                    row.last_error = f"SNMP poll failed: {result}"
+                polls.append(Polled(now, device.id, device.name, failed=f"SNMP poll failed: {result}"))
+                device.last_error = polls[-1].failed
                 continue
-                scan = scans.get(device.id)
-                if scan is not None and scan.error is None:
-                    row.tables, row.tables_at = scan.rows, now
-                row.last_error = result.error
+            p = Polled(now, device.id, device.name, error=result.error, scan=scans.get(device.id),
+                       answered=bool(result.readings or result.texts), missing=list(result.missing))
+            if p.scan is not None and p.scan.error is None:
+                # Kept on the copy too, so a stall doesn't walk it again every poll.
+                device.tables, device.tables_at = p.scan.rows, now
+            device.last_error = result.error
             missing = set(result.missing)
-                if result.readings or result.texts:
-                    row.last_ok_at = now
-                texts = dict(row.texts or {}) if result.error else {}
             for reading in asked[device.id]:
                 value = result.readings.get(reading.key)
                 if reading.unit == "state":
@@ -112,7 +166,7 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
                     raw = value if value is not None else result.texts.get(reading.key)
                     if raw is None:
                         continue
-                        texts[reading.key], value = state_of(reading, raw)
+                    p.texts[reading.key], value = state_of(reading, raw)
                 elif value is None:
                     if reading.key in result.texts:      # a word where a number was asked for
                         missing.add(reading.key)
@@ -121,10 +175,42 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
                     value = RATES.rate(device.id, reading, value, now)
                 else:
                     value = value * reading.scale + reading.offset
-                    if value is None:
+                if value is not None:
+                    p.values.append((reading.key, value))
+            p.missing = sorted(missing)
+            polls.append(p)
+        held.hold(polls)
+    await held.flush(STORE_WAIT_S)
+
+
+async def _store(app: "Quart", rounds: list[list[Polled]]) -> None:
+    """Write held polls, oldest first, in one transaction."""
+    from steward.core.alerts import MetricBatch
+
+    from .models import SnmpDevice
+
+    async with app.db_sessionmaker() as session:
+        async with session.begin():
+            batch = MetricBatch()
+            for polls in rounds:
+                for p in polls:
+                    row = await session.get(SnmpDevice, p.device_id)
+                    if row is None:          # deleted since it was polled
                         continue
-                    batch.add(source_module="snmp", resource_name=device.name,
-                              metric_name=reading.key, value=value, recorded_at=now)
-                row.missing = sorted(missing)
-                row.texts = texts
+                    row.last_polled_at = p.at
+                    if p.failed:
+                        row.last_error = p.failed
+                        continue
+                    if p.scan is not None and p.scan.error is None:
+                        row.tables, row.tables_at = p.scan.rows, p.at
+                    row.last_error = p.error
+                    if p.answered:
+                        row.last_ok_at = p.at
+                    # A failed poll keeps the words it had; one that answered says them afresh.
+                    texts = dict(row.texts or {}) if p.error else {}
+                    texts.update(p.texts)
+                    row.missing, row.texts = p.missing, texts
+                    for key, value in p.values:
+                        batch.add(source_module="snmp", resource_name=p.name,
+                                  metric_name=key, value=value, recorded_at=p.at)
             await batch.record(session)

steward/core/holding.py

@@ -0,0 +1,99 @@
+"""Collected readings held in memory until the database takes them (#5995).
+
+Steward's Postgres sits on the NFS storage it watches. When that storage
+stalls, every write waits, and a collector that stores what it found before
+collecting again stops collecting for as long as the stall lasts: on
+2026-10-17 the NAS readings went blind for the fifteen minutes they mattered.
+
+A collector built on a Holding collects on schedule whatever the database is
+doing, and hands each round to the holding. One writer at a time stores what
+is held, oldest first, each item at the time it was collected, so a stall
+delays the readings instead of losing them, and alert rules count them in the
+order they happened.
+
+    held = holding(app, "SNMP poll", store=_store, limit=120)
+    held.hold(round)                     # one collected round
+    await held.flush(wait_s=20)          # store it, or leave it to the writer
+
+The holding is bounded: past `limit` rounds the oldest is dropped, and that
+is logged with how many have gone. A store has `deadline_s`: generous, past
+any stall seen so far, so a connection that will never answer is given up
+and the rounds stay held for a fresh one rather than waiting forever.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections import deque
+from typing import Awaitable, Callable
+
+logger = logging.getLogger(__name__)
+
+
+class Holding:
+    def __init__(self, name: str, store: Callable[[list], Awaitable[None]], limit: int,
+                 deadline_s: float = 1800.0) -> None:
+        self.name = name
+        self.store = store
+        self.limit = limit
+        self.deadline_s = deadline_s
+        self._held: deque[tuple[int, object]] = deque()
+        self._seq = 0
+        self._writer: asyncio.Task | None = None
+        self.dropped = 0
+
+    def __len__(self) -> int:
+        return len(self._held)
+
+    def hold(self, item) -> None:
+        """Hold one collected round until it is stored."""
+        self._seq += 1
+        self._held.append((self._seq, item))
+        while len(self._held) > self.limit:
+            self._held.popleft()
+            self.dropped += 1
+            logger.warning("%s: the database hasn't taken a round in %d; dropped the oldest "
+                           "(%d dropped so far)", self.name, self.limit, self.dropped)
+
+    async def flush(self, wait_s: float) -> bool:
+        """Start the writer if it isn't running, and wait up to `wait_s` for
+        it. True when nothing is left held. A writer still waiting on the
+        database is left to finish: the next round joins what it stores next."""
+        writer = self._writer
+        if writer is None or writer.done() or writer.get_loop() is not asyncio.get_running_loop():
+            if not self._held:
+                return True
+            writer = self._writer = asyncio.create_task(self._write())
+        done, _ = await asyncio.wait({writer}, timeout=wait_s)
+        if not done:
+            logger.warning("%s: the database hasn't taken %d held round(s) in %.0f s; "
+                           "collecting on", self.name, len(self._held), wait_s)
+        return not self._held
+
+    async def _write(self) -> None:
+        while self._held:
+            batch = list(self._held)
+            try:
+                await asyncio.wait_for(self.store([item for _, item in batch]), self.deadline_s)
+            except TimeoutError:
+                logger.error("%s: storing %d held round(s) took over %.0f s; given up, they stay "
+                             "held (the database may still finish it)", self.name, len(batch),
+                             self.deadline_s)
+                return
+            except Exception:
+                logger.exception("%s: storing %d held round(s) failed; they stay held",
+                                 self.name, len(batch))
+                return
+            last = batch[-1][0]
+            while self._held and self._held[0][0] <= last:
+                self._held.popleft()
+
+
+def holding(app, name: str, store: Callable[[list], Awaitable[None]], limit: int) -> Holding:
+    """The app's Holding for one collector, made on first use."""
+    holdings = getattr(app, "_holdings", None)
+    if holdings is None:
+        holdings = app._holdings = {}
+    if name not in holdings:
+        holdings[name] = Holding(name, store, limit)
+    return holdings[name]

tests/core/test_holding.py

@@ -0,0 +1,64 @@
+"""Collected rounds held in memory until the database takes them (#5995):
+stored oldest first, one writer at a time, bounded, and kept through a
+store that fails or never answers."""
+import asyncio
+from types import SimpleNamespace as NS
+
+from steward.core.holding import Holding, holding
+
+
+def test_rounds_held_while_the_store_hangs_are_stored_in_order_once_it_answers():
+    async def main():
+        gate, stored = asyncio.Event(), []
+
+        async def store(rounds):
+            await gate.wait()
+            stored.append(list(rounds))
+
+        held = Holding("t", store, limit=10)
+        held.hold("a")
+        assert await held.flush(0.05) is False           # the store hangs: the caller goes on
+        held.hold("b")
+        assert await held.flush(0.05) is False and len(held) == 2
+        gate.set()
+        assert await held.flush(1.0) is True and len(held) == 0
+        return stored
+
+    assert asyncio.run(main()) == [["a"], ["b"]]         # "b" waited for the store in flight
+
+
+def test_past_its_limit_the_oldest_round_goes_and_is_counted():
+    held = Holding("t", store=None, limit=2)
+    for r in ("a", "b", "c"):
+        held.hold(r)
+    assert len(held) == 2 and held.dropped == 1 and [i for _, i in held._held] == ["b", "c"]
+
+
+def test_a_store_that_fails_or_overruns_its_deadline_keeps_the_rounds_held():
+    async def main():
+        calls = []
+
+        async def failing(rounds):
+            calls.append(list(rounds))
+            raise RuntimeError("database refused")
+
+        held = Holding("t", failing, limit=10)
+        held.hold("a")
+        assert await held.flush(1.0) is False and len(held) == 1
+
+        async def never(rounds):
+            await asyncio.sleep(3600)
+
+        stuck = Holding("t", never, limit=10, deadline_s=0.05)
+        stuck.hold("a")
+        assert await stuck.flush(1.0) is False and len(stuck) == 1   # given up, not lost
+        return calls
+
+    assert asyncio.run(main()) == [["a"]]
+
+
+def test_an_app_has_one_holding_per_collector():
+    app = NS()
+    first = holding(app, "SNMP poll", store=None, limit=5)
+    assert holding(app, "SNMP poll", store=None, limit=5) is first
+    assert holding(app, "UniFi poll", store=None, limit=5) is not first

tests/plugins/host_agent/test_agent_build_payload.py

@@ -120,11 +120,11 @@ def test_backoff_schedule_stays_inside_the_stale_window():
     assert next_backoff(60) == 60    # capped below Steward's 180 s stale window (#5154)
 
 
-def test_a_backlog_drains_a_few_samples_a_post_oldest_first():
+def test_a_backlog_drains_a_few_samples_a_post_oldest_first_and_the_new_one_waits_its_turn():
     from plugins.host_agent.agent import MAX_SAMPLES_PER_POST, split_backlog
-    buffered = list(range(20))
-    send, held = split_backlog(buffered)
-    assert send == [0, 1, 2, 3] and held == list(range(4, 20))
-    assert len(send) + 1 == MAX_SAMPLES_PER_POST    # room for the new sample
-    assert split_backlog([7]) == ([7], [])
-    assert split_backlog([]) == ([], [])
+    queued = list(range(20)) + ["new"]
+    send, held = split_backlog(queued)
+    assert send == [0, 1, 2, 3, 4] and len(send) == MAX_SAMPLES_PER_POST
+    assert held == list(range(5, 20)) + ["new"]        # never ahead of an older sample
+    assert split_backlog(["new"]) == (["new"], [])
+    assert split_backlog([7, "new"]) == ([7, "new"], [])

tests/plugins/snmp/test_scheduler.py

@@ -0,0 +1,55 @@
+"""An SNMP poll collects on schedule while the database hangs, and what it
+held is stored oldest first, each at its own time, once it answers (#5995)."""
+import asyncio
+from contextlib import asynccontextmanager
+from types import SimpleNamespace as NS
+
+import plugins.snmp.scheduler as scheduler
+import plugins.snmp.service as service
+from plugins.snmp.poller import Credentials, PollResult
+
+
+def _device():
+    return NS(id="d1", name="nas", address="10.0.0.5", port=161, version="2c", presets=["system"],
+              interfaces="", custom=None, tables={}, tables_at=None, last_error=None, texts={})
+
+
+def test_polls_go_on_while_the_database_hangs_and_are_stored_in_order_after(monkeypatch):
+    async def main():
+        stalled = asyncio.Event()
+
+        @asynccontextmanager
+        async def hung_session():
+            await stalled.wait()                         # the storage under Postgres has stalled
+            yield None
+
+        app = NS(db_sessionmaker=hung_session)
+        gate, stored = asyncio.Event(), []
+
+        async def store(app_, rounds):
+            await gate.wait()
+            stored.extend(rounds)
+
+        uptimes = iter([100.0, 200.0])
+
+        async def poll(address, port, creds, oids):
+            return PollResult(readings={"uptime": next(uptimes)})
+
+        monkeypatch.setattr(scheduler, "DEVICES", [_device()])
+        monkeypatch.setattr(scheduler, "DEVICES_READ_S", 0.05)
+        monkeypatch.setattr(scheduler, "STORE_WAIT_S", 0.05)
+        monkeypatch.setattr(scheduler, "_store", store)
+        monkeypatch.setattr(service, "credentials", lambda d: Credentials(community="public"))
+        scheduler.RATES = None
+
+        await scheduler._do_poll(app, poll=poll)         # the device list is the last one read
+        await asyncio.sleep(0.01)
+        await scheduler._do_poll(app, poll=poll)         # the first store hasn't returned: polled anyway
+        gate.set()
+        await app._holdings["SNMP poll"].flush(1.0)
+        return stored
+
+    rounds = asyncio.run(main())
+    assert [[(p.name, p.values) for p in r] for r in rounds] == [
+        [("nas", [("uptime", 100.0)])], [("nas", [("uptime", 200.0)])]]
+    assert rounds[0][0].at < rounds[1][0].at              # each stored at the time it was collected