fix(metrics): seed the latest table in the background, not in the migration (#5208)

Migration 0030 seeded plugin_metric_latest from the whole history: two index
probes per distinct metric key. On the live instance that held the first
boot for 7+ minutes with Steward down. The migration now only creates the
table. Live sources fill it on their next sample, and backfill_latest runs
after startup for resources that have history but no latest rows. Per
resource it does one backward probe for the last sample, then takes the
newest value per metric within the hour before it. Resources that already
have rows are skipped, so later boots only walk the resource names.

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

steward/app.py

@@ -362,10 +362,19 @@ def create_app(
         if not testing:
             await _mark_interrupted_runs(app)
             asyncio.create_task(_run_scheduler(app))
+            app._latest_backfill = asyncio.create_task(_backfill_latest(app))
 
     return app
 
 
+async def _backfill_latest(app: Quart) -> None:
+    from .core.metric_series import backfill_latest
+    try:
+        await backfill_latest(app.db_sessionmaker, app.logger)
+    except Exception:
+        app.logger.exception("latest metrics: backfill failed; quiet resources fill on their next sample")
+
+
 def _register_core_tasks(app: Quart) -> None:
     from .core.scheduler import ScheduledTask
 

steward/core/metric_series.py

@@ -91,6 +91,57 @@ async def latest_by_resource(db, source_module: str, resources, metrics=None,
     return out
 
 
+_BACKFILL_SQL = text("""
+    INSERT INTO plugin_metric_latest
+        (source_module, resource_name, metric_name, value, recorded_at)
+    SELECT DISTINCT ON (metric_name) source_module, resource_name, metric_name,
+           value, recorded_at
+    FROM plugin_metrics
+    WHERE source_module = :src AND resource_name = :res
+      AND recorded_at > :since AND recorded_at <= :last
+    ORDER BY metric_name, recorded_at DESC
+    ON CONFLICT DO NOTHING
+""")
+
+
+async def backfill_latest(sessionmaker, log) -> int:
+    """Fill plugin_metric_latest for resources that have history but no latest
+    row yet: the ones that stopped reporting before the table existed (#5208).
+    A live resource fills itself on its next sample.
+
+    Runs in the background after startup so it never holds the boot. Per
+    resource it costs one backward probe for the last sample and a range over
+    the hour before it; a resource that already has latest rows is skipped, so
+    later boots only step through the resource names. Returns rows written.
+    """
+    import time
+    started, written, filled = time.monotonic(), 0, 0
+    async with sessionmaker() as db:
+        todo = []
+        for src in await sources(db):
+            have = set((await db.execute(
+                select(PluginMetricLatest.resource_name)
+                .where(PluginMetricLatest.source_module == src).distinct()
+            )).scalars())
+            todo += [(src, res) for res in await resources(db, src) if res not in have]
+    for src, res in todo:
+        async with sessionmaker() as db, db.begin():
+            last = (await db.execute(
+                select(func.max(PluginMetric.recorded_at)).where(
+                    PluginMetric.source_module == src, PluginMetric.resource_name == res)
+            )).scalar()
+            if last is None:
+                continue
+            result = await db.execute(_BACKFILL_SQL, {
+                "src": src, "res": res, "last": last, "since": last - timedelta(hours=1)})
+            written += result.rowcount or 0
+            filled += 1
+    if filled:
+        log.info("latest metrics: backfilled %d values for %d quiet resources in %.1fs",
+                 written, filled, time.monotonic() - started)
+    return written
+
+
 async def points(db, source_module: str, resource: str, metric: str, since: datetime,
                  *, target: int = 120) -> list[tuple[datetime, float]]:
     """(time, average) points for one metric since `since`, oldest first.

steward/migrations/versions/0030_plugin_metric_latest.py

@@ -23,32 +23,10 @@ def upgrade() -> None:
         sa.Column("value", sa.Float, nullable=False),
         sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False),
     )
-    # Seed from history without a full scan: step through the distinct keys of
-    # the (source, resource, metric, recorded_at) index one probe at a time,
-    # then take each key's newest row with a backward probe of the same index.
-    op.execute("""
-        WITH RECURSIVE k AS (
-            (SELECT source_module, resource_name, metric_name FROM plugin_metrics
-             ORDER BY source_module, resource_name, metric_name LIMIT 1)
-            UNION ALL
-            SELECT n.source_module, n.resource_name, n.metric_name
-            FROM k, LATERAL (
-                SELECT p.source_module, p.resource_name, p.metric_name FROM plugin_metrics p
-                WHERE (p.source_module, p.resource_name, p.metric_name)
-                    > (k.source_module, k.resource_name, k.metric_name)
-                ORDER BY p.source_module, p.resource_name, p.metric_name LIMIT 1
-            ) n
-        )
-        INSERT INTO plugin_metric_latest
-            (source_module, resource_name, metric_name, value, recorded_at)
-        SELECT k.source_module, k.resource_name, k.metric_name, l.value, l.recorded_at
-        FROM k, LATERAL (
-            SELECT p.value, p.recorded_at FROM plugin_metrics p
-            WHERE p.source_module = k.source_module AND p.resource_name = k.resource_name
-              AND p.metric_name = k.metric_name
-            ORDER BY p.recorded_at DESC LIMIT 1
-        ) l
-    """)
+    # Not seeded here: a seed over the whole history held the first boot for
+    # minutes. Live sources fill it on their next sample, and
+    # metric_series.backfill_latest fills the quiet ones in the background
+    # after startup.
 
 
 def downgrade() -> None:

tests/integration/test_metric_latest.py

@@ -148,3 +148,39 @@ def test_a_renamed_host_carries_its_latest_values(app):
     assert old_latest == {}
     assert new_latest["cpu_pct"][0] == 5.0
     assert new_root["disk_used_pct"][0] == 60.0
+
+
+def test_the_backfill_fills_a_quiet_resource_once_and_leaves_live_ones_alone(app):
+    import logging
+
+    from steward.core.alerts import record_metrics
+    from steward.core.metric_series import backfill_latest
+
+    now = datetime.now(timezone.utc)
+    quiet, live = ("bf-quiet-" + uuid.uuid4().hex[:8]), ("bf-live-" + uuid.uuid4().hex[:8])
+    went_quiet = now - timedelta(hours=3)
+
+    async def go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                # History from before the table existed: rows with no latest.
+                s.add_all([
+                    _sample(quiet, "cpu_pct", 10.0, went_quiet - timedelta(minutes=30)),
+                    _sample(quiet, "cpu_pct", 11.0, went_quiet),
+                    _sample(quiet, "mem_used_pct", 50.0, went_quiet - timedelta(minutes=10)),
+                    _sample(quiet, "load_1m", 9.9, went_quiet - timedelta(hours=2)),  # outside the hour
+                ])
+            async with s.begin():
+                await record_metrics(s, [_sample(live, "cpu_pct", 70.0, now)])
+                s.add(_sample(live, "cpu_pct", 1.0, now - timedelta(minutes=5)))
+        first = await backfill_latest(app.db_sessionmaker, logging.getLogger("test"))
+        second = await backfill_latest(app.db_sessionmaker, logging.getLogger("test"))
+        async with app.db_sessionmaker() as s:
+            return first, second, await _latest(s, quiet), await _latest(s, live)
+
+    first, second, quiet_latest, live_latest = asyncio.run(go())
+    assert first >= 2 and second == 0
+    assert quiet_latest["cpu_pct"][0] == 11.0
+    assert quiet_latest["mem_used_pct"][0] == 50.0
+    assert "load_1m" not in quiet_latest
+    assert live_latest == {"cpu_pct": (70.0, live_latest["cpu_pct"][1])}