perf(monitors): a monitor's latest results by a LATERAL probe, not row_number over its history (#5208)

_last_n_results (Status, the monitors list, host monitors), host_status
(hosts overview, list_hosts) and the report's down-monitor list each
ranked every result a monitor ever recorded to keep the newest few. Each
monitor now gets ORDER BY checked_at DESC LIMIT n in a LATERAL subquery,
which walks ix_monitor_results_monitor_checked backwards and stops.

Also fixes the latest-table test, which read between two begin() blocks
and so autobegan a transaction the second block then refused.

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

steward/core/reports.py

@@ -102,28 +102,13 @@ async def build_report_data(app) -> dict:
         ]
 
         # ── Monitors currently down (latest result is_up = false) ─────────────
-        latest_subq = (
-            select(MonitorResult.monitor_id, func.max(MonitorResult.checked_at).label("max_at"))
-            .group_by(MonitorResult.monitor_id)
-            .subquery()
-        )
-        down_result = await db.execute(
-            select(Monitor.name)
-            .join(MonitorResult, MonitorResult.monitor_id == Monitor.id)
-            .join(
-                latest_subq,
-                and_(
-                    MonitorResult.monitor_id == latest_subq.c.monitor_id,
-                    MonitorResult.checked_at == latest_subq.c.max_at,
-                ),
-            )
-            .where(
-                Monitor.enabled.is_(True),
-                MonitorResult.is_up.is_(False),
-            )
-            .order_by(Monitor.name)
-        )
-        hosts_down = [row.name for row in down_result]
+        from steward.core.status import _last_n_results
+        enabled = (await db.execute(
+            select(Monitor.id, Monitor.name).where(Monitor.enabled.is_(True))
+        )).all()
+        latest = await _last_n_results(db, [m.id for m in enabled], n=1)
+        hosts_down = sorted(name for mid, name in enabled
+                            if latest.get(mid) and not latest[mid][-1].is_up)
 
     return {
         "generated_at": now,

steward/core/status.py

@@ -105,25 +105,33 @@ def sparkline_svg(values: list[float], width: int = 80, height: int = 20,
 
 async def _last_n_results(db, monitor_ids: list[str],
                           n: int = HEARTBEAT_COUNT) -> dict[str, list]:
-    """Return {monitor_id: [MonitorResult]} of the last n results, oldest-first."""
-    from steward.models.monitors import MonitorResult
+    """Return {monitor_id: [MonitorResult]} of the last n results, oldest-first.
+
+    One LATERAL probe per monitor walks ix_monitor_results_monitor_checked
+    backwards and stops after n rows. A row_number() over each monitor's
+    results read its whole history first, which held Status and the monitor
+    widgets at ~16 s (#5208).
+    """
+    from sqlalchemy import true
+    from sqlalchemy.orm import aliased
+    from steward.models.monitors import Monitor, MonitorResult
     if not monitor_ids:
         return {}
-    rn = func.row_number().over(
-        partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc()
-    ).label("rn")
-    subq = select(MonitorResult.id, rn).where(
-        MonitorResult.monitor_id.in_(monitor_ids)
-    ).subquery()
-    res = await db.execute(
+    ids = select(Monitor.id.label("mid")).where(Monitor.id.in_(monitor_ids)).subquery("wanted")
+    recent = (
         select(MonitorResult)
-        .join(subq, MonitorResult.id == subq.c.id)
-        .where(subq.c.rn <= n)
-        .order_by(MonitorResult.monitor_id, MonitorResult.checked_at.asc())
+        .where(MonitorResult.monitor_id == ids.c.mid)
+        .order_by(MonitorResult.checked_at.desc())
+        .limit(n)
+        .lateral("recent")
     )
+    row = aliased(MonitorResult, recent)
+    res = await db.execute(select(row).select_from(ids).join(recent, true()))
     out: dict[str, list] = {mid: [] for mid in monitor_ids}
-    for row in res.scalars():
-        out[row.monitor_id].append(row)
+    for r in res.scalars():
+        out[r.monitor_id].append(r)
+    for rows in out.values():
+        rows.sort(key=lambda r: r.checked_at)
     return out
 
 

steward/hosts/service.py

@@ -93,21 +93,12 @@ async def host_status(db, host_ids: list[str]) -> dict[str, dict]:
     """
     if not host_ids:
         return {}
-    rn = func.row_number().over(
-        partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc()
-    ).label("rn")
-    subq = (
-        select(MonitorResult.id, rn)
-        .join(Monitor, Monitor.id == MonitorResult.monitor_id)
+    monitors = (await db.execute(
+        select(Monitor.id, Monitor.host_id)
         .where(Monitor.host_id.in_(host_ids), Monitor.enabled.is_(True))
-        .subquery()
-    )
-    rows = (await db.execute(
-        select(MonitorResult, Monitor.host_id)
-        .join(subq, MonitorResult.id == subq.c.id)
-        .join(Monitor, Monitor.id == MonitorResult.monitor_id)
-        .where(subq.c.rn == 1)
     )).all()
+    latest = await _last_n_results(db, [m.id for m in monitors], n=1)
+    rows = [(latest[mid][-1], host_id) for mid, host_id in monitors if latest.get(mid)]
     out: dict[str, dict] = {}
     for res, host_id in rows:
         d = out.setdefault(host_id, {"state": "pending", "latency_ms": None, "count": 0})

tests/integration/test_metric_latest.py

@@ -54,7 +54,8 @@ def test_the_newest_sample_wins_and_a_late_backlog_does_not_move_it_back(app):
                     _sample(host, "cpu_pct", 20.0, now - timedelta(minutes=1)),
                     _sample(host, "mem_used_pct", 40.0, now - timedelta(minutes=1)),
                 ])
-            after_first = await _latest(s, host)
+            async with s.begin():
+                after_first = await _latest(s, host)
             async with s.begin():
                 # An agent catching up posts older samples after newer ones.
                 await record_metrics(s, [

tests/integration/test_monitors.py

@@ -83,3 +83,45 @@ def test_status_source_rolls_up_results(app):
     assert e.status == "up"                 # latest result is up
     assert len(e.heartbeat) == 4
     assert e.uptime["24h"] == 75.0          # 3 of 4 up
+
+
+@_NEEDS_DB
+def test_last_results_are_the_newest_n_oldest_first_and_host_status_reads_the_last(app):
+    """_last_n_results takes each monitor's newest n by a LATERAL probe (#5208);
+    host_status reads the newest one per enabled monitor."""
+    from steward.core.status import _last_n_results
+    from steward.hosts.service import host_status
+    from steward.models.hosts import Host
+    from steward.models.monitors import Monitor, MonitorResult
+
+    now = datetime.now(timezone.utc)
+    hid, a, b, quiet = (str(uuid.uuid4()) for _ in range(4))
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(Host(id=hid, name="lateral-" + hid[:8], address="10.8.8.8"))
+            async with s.begin():
+                s.add(Monitor(id=a, name="a-" + a[:8], type="icmp", target="10.8.8.8",
+                              host_id=hid, config_json="{}", enabled=True))
+                s.add(Monitor(id=b, name="b-" + b[:8], type="icmp", target="10.8.8.9",
+                              host_id=hid, config_json="{}", enabled=True))
+                s.add(Monitor(id=quiet, name="q-" + quiet[:8], type="icmp", target="10.8.8.7",
+                              config_json="{}", enabled=True))
+            async with s.begin():
+                for i in range(5):    # a: up, up, up, up, down (newest)
+                    s.add(MonitorResult(monitor_id=a, checked_at=now - timedelta(minutes=5 - i),
+                                        is_up=i < 4))
+                s.add(MonitorResult(monitor_id=b, checked_at=now - timedelta(minutes=1),
+                                    is_up=True, response_ms=12.0))
+            async with s.begin():
+                last = await _last_n_results(s, [a, b, quiet], n=3)
+                status = await host_status(s, [hid])
+            return last, status
+
+    last, status = asyncio.run(_go())
+    assert [r.is_up for r in last[a]] == [True, True, False]
+    assert last[a][0].checked_at < last[a][-1].checked_at       # oldest first
+    assert len(last[b]) == 1 and last[quiet] == []
+    assert status[hid]["state"] == "down" and status[hid]["count"] == 2
+    assert status[hid]["latency_ms"] == 12.0