fix(monitors): run each monitor on its own interval, from its scheduled start (#5151)

The dispatcher ticked once per MONITORS_POLL_INTERVAL and stamped
last_checked_at when the whole batch finished. With a 60 s tick and a 60 s
monitor the next tick saw 59.x s elapsed and skipped, so the monitor ran
every 120 s; a 30 s monitor could never run faster than the tick.

- The dispatcher ticks every 5 s (DISPATCH_TICK_SECONDS) and reads the
  monitors and intervals fresh each tick, so a changed interval applies at
  the next one.
- Due is measured from the scheduled start, stamped before the check runs,
  with half a tick of slack so tick jitter never costs a whole tick. A
  long-overdue monitor runs once.
- Each check runs in its own task and records its own result, so a slow
  probe holds up nobody; a monitor still being checked is not started
  again, and it is freed in a finally.
- A check past CHECK_DEADLINE_SECONDS (120) is recorded down with "Check did
  not finish within 120 s" rather than holding its monitor forever.

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

steward/app.py

@@ -367,7 +367,7 @@ def create_app(
 def _register_core_tasks(app: Quart) -> None:
     from .core.scheduler import ScheduledTask
 
-    poll_interval = app.config.get("MONITORS_POLL_INTERVAL", 60)
+    from .monitors.scheduler import DISPATCH_TICK_SECONDS as MONITOR_DISPATCH_TICK
     cleanup_interval = 3600  # hourly
 
     async def run_monitor_checks():
@@ -418,10 +418,12 @@ def _register_core_tasks(app: Quart) -> None:
             interval_seconds=60,
             run_on_startup=False,
         ),
+        # Ticks often and only starts checks: each monitor decides whether it
+        # is due from its own interval (steward/monitors/scheduler.py).
         ScheduledTask(
             name="monitor_check",
             coro_factory=run_monitor_checks,
-            interval_seconds=poll_interval,
+            interval_seconds=MONITOR_DISPATCH_TICK,
             run_on_startup=True,
         ),
         ScheduledTask(

steward/monitors/scheduler.py

@@ -1,52 +1,92 @@
-"""Single scheduled task that runs every due Monitor of any type."""
+"""Dispatch every due Monitor of any type, each check in its own task.
+
+The dispatcher ticks every DISPATCH_TICK_SECONDS. Each tick it reads the
+enabled monitors and their intervals fresh (a changed interval applies at the
+next tick), stamps the due ones with the time they were due to start, and
+starts each check on its own. A slow check (a 10 s timeout) holds up nobody
+else, and a monitor still being checked is not started twice.
+"""
 from __future__ import annotations
 import asyncio
 import logging
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 
 from steward.core.alerts import MetricBatch
 from steward.models.monitors import Monitor, MonitorResult
-from .runner import run_monitor
+from .runner import _BLANK, run_monitor
 
 logger = logging.getLogger(__name__)
 
+# How often the dispatcher looks for due monitors. A monitor runs within one
+# tick of its interval, so this is the cadence's worst-case lateness.
+DISPATCH_TICK_SECONDS = 5
+# A check past this is recorded down rather than left holding its monitor:
+# far beyond any probe's own timeout, so it only ever catches a hang.
+CHECK_DEADLINE_SECONDS = 120
+
+
+def is_due(monitor, now: datetime, global_interval: int) -> bool:
+    """Whether a monitor's interval has elapsed since its last scheduled start.
+
+    Half a tick of slack, so a check stamped at t isn't missed at t+interval
+    by a few milliseconds of tick jitter and pushed back a whole tick."""
+    if monitor.last_checked_at is None:
+        return True
+    interval = monitor.check_interval_seconds or global_interval
+    elapsed = (now - monitor.last_checked_at).total_seconds()
+    return elapsed >= interval - DISPATCH_TICK_SECONDS / 2
+
+
+def _state(app) -> tuple[set, set]:
+    """(monitor ids being checked, their tasks) for this app."""
+    if not hasattr(app, "_monitor_checks"):
+        app._monitor_checks = (set(), set())
+    return app._monitor_checks
+
 
 async def run_due_monitors(app) -> None:
-    """Check every enabled Monitor whose interval has elapsed, persist results.
+    """Start a check for every enabled Monitor that is due and not already
+    being checked. Returns once they are started, not finished.
 
     Per-monitor `check_interval_seconds` overrides the global
-    MONITORS_POLL_INTERVAL (0 = use global), mirroring how the scheduler fires
-    on the global cadence but each monitor decides if it is actually due.
-    """
+    MONITORS_POLL_INTERVAL (0 = use global)."""
     global_interval = int(app.config.get("MONITORS_POLL_INTERVAL", 60))
     now = datetime.now(timezone.utc)
+    in_flight, tasks = _state(app)
 
     async with app.db_sessionmaker() as session:
         monitors = list((await session.execute(
             select(Monitor).where(Monitor.enabled.is_(True))
         )).scalars())
+        due = [m for m in monitors if m.id not in in_flight and is_due(m, now, global_interval)]
+        if not due:
+            return
+        # Stamp the scheduled start before checking: the next run is due one
+        # interval from now, however long this check takes.
+        await session.execute(
+            update(Monitor).where(Monitor.id.in_([m.id for m in due])).values(last_checked_at=now))
+        await session.commit()
 
-    due: list[Monitor] = []
-    for m in monitors:
-        interval = m.check_interval_seconds or global_interval
-        if m.last_checked_at is None or (now - m.last_checked_at).total_seconds() >= interval:
-            due.append(m)
-    if not due:
-        return
+    for m in due:
+        in_flight.add(m.id)
+        task = asyncio.create_task(_check_and_record(app, m))
+        tasks.add(task)
+        task.add_done_callback(tasks.discard)
 
-    pairs = await asyncio.gather(
-        *[run_monitor(m) for m in due], return_exceptions=True
-    )
-    check_time = datetime.now(timezone.utc)
 
-    async with app.db_sessionmaker() as session:
-        async with session.begin():
-            for monitor, res in zip(due, pairs):
-                if isinstance(res, Exception):
-                    logger.error("Monitor %r check errored: %s", monitor.name, res)
-                    continue
+async def _check_and_record(app, monitor: Monitor) -> None:
+    """Run one monitor's check and store its result and metrics."""
+    in_flight, _ = _state(app)
+    try:
+        try:
+            res = await asyncio.wait_for(run_monitor(monitor), CHECK_DEADLINE_SECONDS)
+        except asyncio.TimeoutError:
+            res = dict(_BLANK, error_msg=f"Check did not finish within {CHECK_DEADLINE_SECONDS} s")
+        check_time = datetime.now(timezone.utc)
+        async with app.db_sessionmaker() as session:
+            async with session.begin():
                 session.add(MonitorResult(
                     monitor_id=monitor.id,
                     checked_at=check_time,
@@ -58,10 +98,6 @@ async def run_due_monitors(app) -> None:
                     tls_expires_at=res["tls_expires_at"],
                     error_msg=res["error_msg"],
                 ))
-                db_monitor = await session.get(Monitor, monitor.id)
-                if db_monitor:
-                    db_monitor.last_checked_at = check_time
-
                 # Alert pipeline — keyed by monitor type (source) + name.
                 batch = MetricBatch()
                 batch.add(
@@ -76,3 +112,7 @@ async def run_due_monitors(app) -> None:
                         value=res["response_ms"],
                     )
                 await batch.record(session)
+    except Exception:
+        logger.exception("Monitor %r check errored", monitor.name)
+    finally:
+        in_flight.discard(monitor.id)

tests/core/test_monitor_schedule.py

@@ -0,0 +1,154 @@
+"""Monitors run on their own interval, not the dispatcher's: due is measured
+from the scheduled start, each check runs on its own, one still running is not
+started twice, and a check that never finishes is recorded down (#5151)."""
+import asyncio
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace as NS
+
+import pytest
+
+from steward.monitors import scheduler as sched
+from steward.monitors.scheduler import DISPATCH_TICK_SECONDS, is_due
+
+NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)
+
+
+def _m(mid="m1", interval=0, checked_ago=None, name=None):
+    return NS(id=mid, name=name or mid, type="http", enabled=True, check_interval_seconds=interval,
+              last_checked_at=None if checked_ago is None else NOW - timedelta(seconds=checked_ago))
+
+
+@pytest.mark.parametrize("monitor, due", [
+    (_m(), True),                                   # never checked
+    (_m(checked_ago=60), True),                     # exactly one interval (global 60)
+    (_m(checked_ago=60 - DISPATCH_TICK_SECONDS / 2), True),  # within half a tick: not pushed a tick late
+    (_m(checked_ago=55), False),
+    (_m(interval=30, checked_ago=30), True),        # its own interval, not the global one
+    (_m(interval=30, checked_ago=20), False),
+    (_m(interval=300, checked_ago=120), False),
+    (_m(checked_ago=86400), True),                  # long overdue: due, and runs once
+])
+def test_is_due(monitor, due):
+    assert is_due(monitor, NOW, 60) is due
+
+
+class _Store:
+    def __init__(self, monitors):
+        self.monitors, self.stamped, self.added = monitors, [], []
+
+
+class _Session:
+    def __init__(self, store):
+        self.store = store
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    def begin(self):
+        return self
+
+    async def execute(self, stmt):
+        if getattr(stmt, "is_update", False):
+            # Like the real stamp: every monitor in this fake was due.
+            self.store.stamped.append(stmt)
+            for m in self.store.monitors:
+                m.last_checked_at = datetime.now(timezone.utc)
+            return NS()
+        return NS(scalars=lambda: iter(self.store.monitors))
+
+    async def commit(self):
+        pass
+
+    def add(self, obj):
+        self.store.added.append(obj)
+
+
+class _Batch:
+    def add(self, **kw):
+        pass
+
+    async def record(self, session):
+        pass
+
+
+def _app(store):
+    return NS(config={"MONITORS_POLL_INTERVAL": 60}, db_sessionmaker=lambda: _Session(store))
+
+
+async def _settle(app):
+    _, tasks = sched._state(app)
+    if tasks:
+        await asyncio.gather(*list(tasks))
+
+
+async def test_a_slow_check_holds_up_nobody_and_is_not_started_twice(monkeypatch):
+    release = asyncio.Event()
+    started = []
+
+    async def fake_run(m):
+        started.append(m.id)
+        if m.id == "slow":
+            await release.wait()
+        return dict(sched._BLANK, is_up=True)
+
+    monkeypatch.setattr(sched, "run_monitor", fake_run)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    store = _Store([_m("slow"), _m("fast")])
+    app = _app(store)
+
+    await sched.run_due_monitors(app)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert sorted(started) == ["fast", "slow"]
+    assert [r.monitor_id for r in store.added] == ["fast"]   # recorded while slow still runs
+
+    await sched.run_due_monitors(app)                          # next tick: slow is in flight
+    await asyncio.sleep(0)
+    assert started.count("slow") == 1
+
+    release.set()
+    await _settle(app)
+    assert sorted(r.monitor_id for r in store.added) == ["fast", "slow"]
+    assert sched._state(app)[0] == set()
+    assert len(store.stamped) == 1   # stamped once, when both were due
+
+
+async def test_a_check_that_raises_frees_its_monitor(monkeypatch):
+    async def boom(m):
+        raise RuntimeError("probe broke")
+
+    monkeypatch.setattr(sched, "run_monitor", boom)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    store = _Store([_m("m1")])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    await _settle(app)
+    assert store.added == [] and sched._state(app)[0] == set()
+
+
+async def test_a_check_that_never_finishes_is_recorded_down(monkeypatch):
+    async def hang(m):
+        await asyncio.sleep(3600)
+
+    monkeypatch.setattr(sched, "run_monitor", hang)
+    monkeypatch.setattr(sched, "MetricBatch", _Batch)
+    monkeypatch.setattr(sched, "CHECK_DEADLINE_SECONDS", 0.01)
+    store = _Store([_m("m1")])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    await _settle(app)
+    [res] = store.added
+    assert res.is_up is False and "did not finish" in res.error_msg
+    assert sched._state(app)[0] == set()
+
+
+async def test_nothing_due_stamps_nothing(monkeypatch):
+    m = _m("m1")
+    m.last_checked_at = datetime.now(timezone.utc) - timedelta(seconds=10)
+    store = _Store([m])
+    app = _app(store)
+    await sched.run_due_monitors(app)
+    assert store.stamped == [] and store.added == []