feat(reliability): fd-leak rails — self-watchdog, poll-overlap guard, fd tests

Guardrails so the fd-leak class of bug (Errno 24 lockup; recent SNMP + UniFi
fixes) surfaces early or can't compound, instead of silently killing the app.

- Self-fd watchdog (steward/core/self_monitor.py): records open_fds and
  open_fds_pct (% of soft RLIMIT_NOFILE) as "steward"/"process" metrics each
  minute through the normal alert pipeline, so the operator can alert on them
  via the existing alert-rules UI. Built-in WARNING floor at 80% gives a
  zero-config early signal. Stdlib-only (/proc + resource); degrades to a no-op
  off Linux. Registered as a core ScheduledTask in app.py.

- Poll-overlap guard (steward/core/scheduler.py): extract a pure _DueTracker
  that skips a tick while a task's prior run is still in flight, so a hung poll
  can't stack overlapping runs (which amplify per-poll resource/fd use). A
  skipped task isn't penalised — it retries the next tick after it completes.

- fd-stability tests (tests/core/): _DueTracker overlap policy, the watchdog
  metric/warning/degradation paths, and a real-fd canary that hammers tcp_check
  and asserts /proc/self/fd doesn't grow.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

steward/app.py

@@ -268,6 +268,10 @@ def _register_core_tasks(app: Quart) -> None:
         from .core.reports import check_and_send
         await check_and_send(app)
 
+    async def run_self_monitor():
+        from .core.self_monitor import record_self_metrics
+        await record_self_metrics(app)
+
     app._task_registry.extend([
         ScheduledTask(
             name="weekly_report_check",
@@ -275,6 +279,15 @@ def _register_core_tasks(app: Quart) -> None:
             interval_seconds=3600,
             run_on_startup=False,
         ),
+        # Watch our own fd usage so a descriptor leak surfaces as a metric/alert
+        # instead of a silent Errno 24 lockup. Cheap; plugin_metrics roll up
+        # hourly so the 60s cadence is storage-bounded.
+        ScheduledTask(
+            name="self_monitor",
+            coro_factory=run_self_monitor,
+            interval_seconds=60,
+            run_on_startup=True,
+        ),
         ScheduledTask(
             name="monitor_check",
             coro_factory=run_monitor_checks,

steward/core/scheduler.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 import asyncio
 import logging
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Callable, Coroutine
 
 logger = logging.getLogger(__name__)
@@ -15,28 +15,65 @@ class ScheduledTask:
     run_on_startup: bool = False
 
 
+@dataclass
+class _DueTracker:
+    """Decides which scheduled tasks are due, with a self-overlap guard.
+
+    Pure (no asyncio, no clock of its own — `now` is passed in) so the
+    scheduling policy is unit-testable without timing races. A task whose prior
+    run is still in flight is NOT re-fired: overlapping poll runs stack up open
+    connections/subprocesses and amplify any per-poll resource use — the same
+    failure mode behind the fd-leak lockups. The skipped task is retried on the
+    next tick once it completes (its last_run isn't advanced while skipped).
+    """
+    last_run: dict[str, float] = field(default_factory=dict)
+    in_flight: set[str] = field(default_factory=set)
+
+    def due(self, tasks: list[ScheduledTask], now: float) -> list[ScheduledTask]:
+        ready: list[ScheduledTask] = []
+        for task in tasks:
+            if now - self.last_run.get(task.name, 0) < task.interval_seconds:
+                continue
+            if task.name in self.in_flight:
+                logger.warning(
+                    "Scheduled task %r still running — skipping this tick",
+                    task.name)
+                continue
+            ready.append(task)
+        return ready
+
+    def mark_started(self, task: ScheduledTask, now: float) -> None:
+        self.in_flight.add(task.name)
+        self.last_run[task.name] = now
+
+    def mark_done(self, name: str) -> None:
+        self.in_flight.discard(name)
+
+
 async def start_scheduler(tasks: list[ScheduledTask]) -> None:
     """Run scheduled tasks in a loop. Call with asyncio.create_task()."""
-    last_run: dict[str, float] = {}
+    tracker = _DueTracker()
+
+    def _spawn(task: ScheduledTask, now: float) -> None:
+        tracker.mark_started(task, now)
+        asyncio.create_task(_run_task(task, tracker))
 
     for task in tasks:
         if task.run_on_startup:
             logger.info(f"Startup task: {task.name}")
-            asyncio.create_task(_run_task(task))
+            _spawn(task, asyncio.get_event_loop().time())
-            last_run[task.name] = asyncio.get_event_loop().time()
 
     while True:
         now = asyncio.get_event_loop().time()
-        for task in tasks:
+        for task in tracker.due(tasks, now):
-            last = last_run.get(task.name, 0)
+            _spawn(task, now)
-            if now - last >= task.interval_seconds:
-                asyncio.create_task(_run_task(task))
-                last_run[task.name] = now
         await asyncio.sleep(1)
 
 
-async def _run_task(task: ScheduledTask) -> None:
+async def _run_task(task: ScheduledTask, tracker: _DueTracker) -> None:
     try:
         await task.coro_factory()
     except Exception:
         logger.exception(f"Scheduled task {task.name!r} raised an exception")
+    finally:
+        tracker.mark_done(task.name)

steward/core/self_monitor.py

@@ -0,0 +1,116 @@
+"""Self-monitoring: record Steward's own open file-descriptor usage.
+
+A leaked socket/file handle in a poll loop (see the SNMP and UniFi fd-leak
+issues) used to fail silently until the process hit its fd ceiling and
+`socket.accept()` started raising `OSError: [Errno 24] Too many open files` —
+taking the whole app down with no early warning.
+
+This turns that failure mode into an observable signal. Steward monitors other
+things; it should monitor itself. We record two metrics each tick under
+`source_module="steward"`:
+
+  • ``open_fds``      — raw count of open descriptors
+  • ``open_fds_pct``  — that count as a percentage of the soft RLIMIT_NOFILE
+
+Both flow through the normal alert pipeline, so the operator can attach an alert
+rule to either via the existing alert-rules UI. As a zero-config floor we also
+log a WARNING once usage crosses ``FD_WARN_PCT`` — a leak becomes visible even
+before any rule is set up.
+"""
+from __future__ import annotations
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+# Stdlib-only fd counting via /proc keeps this dependency-free. resource is
+# POSIX-only but always present on the Linux runtime image; guard anyway so
+# imports never explode on a dev machine.
+try:
+    import resource
+except ImportError:  # pragma: no cover - non-POSIX
+    resource = None  # type: ignore[assignment]
+
+# Warn (without needing a configured alert rule) once we're using this fraction
+# of the soft fd limit. 80% leaves headroom to act before accepts start failing.
+FD_WARN_PCT = 80.0
+
+_warned = False  # de-dupe the WARNING so a sustained leak doesn't spam the log
+
+
+def count_open_fds() -> int | None:
+    """Number of open file descriptors for this process, or None if unknown.
+
+    Reads ``/proc/self/fd`` (Linux). Returns None where /proc isn't available
+    (e.g. a macOS dev box) so callers degrade to a no-op rather than guessing.
+    """
+    try:
+        return len(os.listdir("/proc/self/fd"))
+    except OSError:
+        return None
+
+
+def fd_soft_limit() -> int | None:
+    """Soft RLIMIT_NOFILE for this process, or None if it can't be read.
+
+    None when the limit is unknown or 'unlimited' (RLIM_INFINITY) — a percentage
+    against an unbounded ceiling is meaningless, so we skip the pct metric then.
+    """
+    if resource is None:
+        return None
+    try:
+        soft, _hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+    except (ValueError, OSError):
+        return None
+    if soft <= 0 or soft == resource.RLIM_INFINITY:
+        return None
+    return soft
+
+
+async def record_self_metrics(app) -> None:
+    """Record open-fd usage as Steward's own metrics; warn past the floor.
+
+    No-op (logged at debug) when fd accounting isn't available on this platform,
+    so it's safe to schedule unconditionally.
+    """
+    global _warned
+
+    fds = count_open_fds()
+    if fds is None:
+        logger.debug("self_monitor: /proc/self/fd unavailable — skipping fd metrics")
+        return
+
+    from .alerts import record_metric
+
+    soft = fd_soft_limit()
+    pct = (fds / soft * 100.0) if soft else None
+
+    async with app.db_sessionmaker() as session:
+        async with session.begin():
+            await record_metric(
+                session=session,
+                source_module="steward",
+                resource_name="process",
+                metric_name="open_fds",
+                value=float(fds),
+            )
+            if pct is not None:
+                await record_metric(
+                    session=session,
+                    source_module="steward",
+                    resource_name="process",
+                    metric_name="open_fds_pct",
+                    value=pct,
+                )
+
+    if pct is not None and pct >= FD_WARN_PCT:
+        if not _warned:
+            logger.warning(
+                "Open file descriptors at %.0f%% of the soft limit (%d/%d) — "
+                "possible descriptor leak; the app will stop accepting "
+                "connections if this reaches 100%%.", pct, fds, soft)
+            _warned = True
+    else:
+        _warned = False  # recovered — re-arm the warning for the next breach
+
+    logger.debug("self_monitor: open_fds=%d soft_limit=%s pct=%s", fds, soft, pct)

tests/core/test_fd_stability.py

@@ -0,0 +1,46 @@
+"""Real-fd regression canary.
+
+Drives a real socket code path (the TCP reachability probe) in a tight loop and
+asserts the process's open-fd count doesn't grow. If someone reintroduces a
+socket leak in the probe path — the class of bug behind the Errno 24 lockups —
+this fails in CI instead of in production. Uses real descriptors, not fakes, so
+the assertion has teeth.
+"""
+import asyncio
+import os
+
+import pytest
+
+from steward.monitors.ping import tcp_check
+
+_ITERATIONS = 100
+
+
+def _fd_count() -> int | None:
+    try:
+        return len(os.listdir("/proc/self/fd"))
+    except OSError:
+        return None
+
+
+async def _hammer_tcp_check() -> None:
+    # 127.0.0.1:1 has no listener → connection refused immediately. Each call must
+    # fully release its socket; a leak would add ~one fd per iteration.
+    for _ in range(_ITERATIONS):
+        await tcp_check("127.0.0.1", 1)
+
+
+def test_tcp_check_does_not_leak_fds():
+    if _fd_count() is None:
+        pytest.skip("/proc/self/fd unavailable on this platform")
+
+    asyncio.run(_hammer_tcp_check())   # warm up (lazy imports, caches)
+    before = _fd_count()
+    asyncio.run(_hammer_tcp_check())
+    after = _fd_count()
+
+    # Small slack for interpreter-internal fds; a genuine leak over 100 iterations
+    # would be far larger than this.
+    assert after - before <= 5, (
+        f"open fds grew {before}->{after} over {_ITERATIONS} tcp_check calls "
+        "— possible socket leak")

tests/core/test_scheduler_overlap.py

@@ -0,0 +1,58 @@
+"""Unit tests for the scheduler's poll-overlap guard (_DueTracker).
+
+The tracker is pure (clock passed in) so we can assert the overlap policy
+deterministically: a task whose prior run is still in flight is never re-fired,
+and a skipped task isn't penalised — it runs on the next tick once it finishes.
+This is the rail that stops a hung poll from stacking overlapping runs (which
+would amplify any per-poll resource/fd use).
+"""
+from steward.core.scheduler import ScheduledTask, _DueTracker
+
+
+def _task(name: str, interval: int) -> ScheduledTask:
+    return ScheduledTask(name=name, coro_factory=lambda: None, interval_seconds=interval)
+
+
+def test_due_only_after_interval_elapses():
+    t = _task("a", 60)
+    tr = _DueTracker()
+    assert tr.due([t], now=59) == []      # 59 < 60
+    assert tr.due([t], now=60) == [t]     # interval elapsed
+
+
+def test_in_flight_task_is_skipped_even_when_due():
+    t = _task("a", 0)  # due every tick
+    tr = _DueTracker()
+    tr.mark_started(t, now=0)
+    assert tr.due([t], now=100) == []     # still running → skipped
+    tr.mark_done("a")
+    assert tr.due([t], now=100) == [t]    # finished → eligible again
+
+
+def test_skip_does_not_advance_last_run_so_it_retries():
+    t = _task("a", 10)
+    tr = _DueTracker()
+    tr.mark_started(t, now=0)
+    assert tr.due([t], now=100) == []     # due but in flight
+    assert tr.last_run["a"] == 0          # not advanced by the skip
+    tr.mark_done("a")
+    assert tr.due([t], now=100) == [t]    # retried promptly after completion
+
+
+def test_mark_started_advances_last_run_and_marks_in_flight():
+    t = _task("a", 10)
+    tr = _DueTracker()
+    tr.mark_started(t, now=50)
+    assert tr.last_run["a"] == 50
+    assert tr.in_flight == {"a"}
+    assert tr.due([t], now=55) == []      # 5 < 10 (not yet due)
+    assert tr.due([t], now=61) == []      # due by interval, but still in flight
+    tr.mark_done("a")
+    assert tr.due([t], now=61) == [t]
+
+
+def test_independent_tasks_do_not_block_each_other():
+    a, b = _task("a", 0), _task("b", 0)
+    tr = _DueTracker()
+    tr.mark_started(a, now=0)             # a hangs
+    assert tr.due([a, b], now=10) == [b]  # b still fires

tests/core/test_self_monitor.py

@@ -0,0 +1,115 @@
+"""Unit tests for the self-fd watchdog (no DB, no network).
+
+Exercises the fd accounting, the percentage/warning floor, and that the
+recorder degrades to a no-op where fd accounting isn't available. The DB session
+and the alert pipeline are faked — we only assert that the right metrics get
+handed to record_metric.
+"""
+import asyncio
+import logging
+
+import steward.core.alerts as alerts
+import steward.core.self_monitor as sm
+
+
+# ── fd accounting ────────────────────────────────────────────────────────────
+
+def test_count_open_fds_sane_or_none():
+    n = sm.count_open_fds()
+    # On the Linux CI image /proc exists; stdin/stdout/stderr are always open.
+    assert n is None or n >= 3
+
+
+def test_fd_soft_limit_positive_or_none():
+    soft = sm.fd_soft_limit()
+    assert soft is None or soft > 0
+
+
+# ── fake session / app plumbing ──────────────────────────────────────────────
+
+class _Ctx:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return False
+
+
+class _FakeSession(_Ctx):
+    def begin(self):
+        return _Ctx()
+
+
+class _FakeApp:
+    def db_sessionmaker(self):
+        return _FakeSession()
+
+
+def _patch_recorder(monkeypatch):
+    recorded: list[tuple] = []
+
+    async def fake_record_metric(session, source_module, resource_name, metric_name, value):
+        recorded.append((source_module, resource_name, metric_name, value))
+
+    monkeypatch.setattr(alerts, "record_metric", fake_record_metric)
+    return recorded
+
+
+# ── record_self_metrics ──────────────────────────────────────────────────────
+
+def test_records_both_fd_metrics(monkeypatch):
+    recorded = _patch_recorder(monkeypatch)
+    monkeypatch.setattr(sm, "count_open_fds", lambda: 50)
+    monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000)
+
+    asyncio.run(sm.record_self_metrics(_FakeApp()))
+
+    assert ("steward", "process", "open_fds", 50.0) in recorded
+    assert ("steward", "process", "open_fds_pct", 5.0) in recorded
+
+
+def test_no_pct_metric_when_limit_unknown(monkeypatch):
+    recorded = _patch_recorder(monkeypatch)
+    monkeypatch.setattr(sm, "count_open_fds", lambda: 50)
+    monkeypatch.setattr(sm, "fd_soft_limit", lambda: None)
+
+    asyncio.run(sm.record_self_metrics(_FakeApp()))
+
+    metric_names = [m for (_s, _r, m, _v) in recorded]
+    assert "open_fds" in metric_names
+    assert "open_fds_pct" not in metric_names   # meaningless without a ceiling
+
+
+def test_warns_past_floor(monkeypatch, caplog):
+    _patch_recorder(monkeypatch)
+    monkeypatch.setattr(sm, "count_open_fds", lambda: 900)
+    monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000)   # 90% ≥ FD_WARN_PCT
+    monkeypatch.setattr(sm, "_warned", False)
+
+    with caplog.at_level(logging.WARNING, logger=sm.logger.name):
+        asyncio.run(sm.record_self_metrics(_FakeApp()))
+
+    assert any("soft limit" in r.message for r in caplog.records)
+
+
+def test_no_warning_below_floor(monkeypatch, caplog):
+    _patch_recorder(monkeypatch)
+    monkeypatch.setattr(sm, "count_open_fds", lambda: 100)
+    monkeypatch.setattr(sm, "fd_soft_limit", lambda: 1000)   # 10%
+    monkeypatch.setattr(sm, "_warned", False)
+
+    with caplog.at_level(logging.WARNING, logger=sm.logger.name):
+        asyncio.run(sm.record_self_metrics(_FakeApp()))
+
+    assert not any("soft limit" in r.message for r in caplog.records)
+
+
+def test_noop_when_fd_count_unavailable(monkeypatch):
+    # /proc absent → must return before touching the DB.
+    monkeypatch.setattr(sm, "count_open_fds", lambda: None)
+
+    class _Boom:
+        def db_sessionmaker(self):
+            raise AssertionError("must not open a session when fds are unknown")
+
+    asyncio.run(sm.record_self_metrics(_Boom()))   # no raise