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>

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,67 +126,91 @@ 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 = {}, {}
 
-    started = datetime.now(timezone.utc)
-    scans, asked = {}, {}
+        async def one(d):
+            creds = credentials(d)
+            if tables_due(d, started):
+                scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or []))
+            found = scans.get(d.id)
+            asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None)
+            return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]])
 
-    async def one(d):
-        creds = credentials(d)
-        if tables_due(d, started):
-            scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or []))
-        found = scans.get(d.id)
-        asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None)
-        return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]])
+        results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True)
 
-    results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True)
+        now = datetime.now(timezone.utc)
+        polls = []
+        for device, result in zip(devices, results):
+            if isinstance(result, BaseException):
+                logger.warning("SNMP poll of %s failed: %s", device.name, result)
+                polls.append(Polled(now, device.id, device.name, failed=f"SNMP poll failed: {result}"))
+                device.last_error = polls[-1].failed
+                continue
+            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)
+            for reading in asked[device.id]:
+                value = result.readings.get(reading.key)
+                if reading.unit == "state":
+                    # Its word is shown as answered; its level is stored,
+                    # unless Steward doesn't know the word (level None).
+                    raw = value if value is not None else result.texts.get(reading.key)
+                    if raw is None:
+                        continue
+                    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)
+                    continue
+                elif reading.counter:
+                    value = RATES.rate(device.id, reading, value, now)
+                else:
+                    value = value * reading.scale + reading.offset
+                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
 
-    now = datetime.now(timezone.utc)
     async with app.db_sessionmaker() as session:
         async with session.begin():
             batch = MetricBatch()
-            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}"
-                    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
-                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":
-                        # Its word is shown as answered; its level is stored,
-                        # unless Steward doesn't know the word (level 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)
-                    elif value is None:
-                        if reading.key in result.texts:      # a word where a number was asked for
-                            missing.add(reading.key)
+            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
-                    elif reading.counter:
-                        value = RATES.rate(device.id, reading, value, now)
-                    else:
-                        value = value * reading.scale + reading.offset
-                    if value is None:
+                    row.last_polled_at = p.at
+                    if p.failed:
+                        row.last_error = p.failed
                         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
+                    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)