materialize step 7: freshness alerting + collapsed-bench ttl keep-alive

src/app/api/cron/health-check/route.ts

@@ -2,6 +2,7 @@ import { timingSafeEqual } from "node:crypto";
 import { NextResponse, type NextRequest } from "next/server";
 import { getBenchmarkSlugs } from "@/data/benchmarks";
 import { getSpecs, loadAllBenchmarks } from "@/lib/spec";
+import { readHeartbeat, storeConfigured } from "@/lib/materialize/store";
 import { extractMetricName, Prometheus } from "@/lib/prometheus";
 
 export const runtime = "nodejs";
@@ -214,6 +215,37 @@ export async function GET(req: NextRequest) {
     }
   }
 
+  // Materialization freshness watchdog. The worker SETs a heartbeat key
+  // after every tier-A sweep (60s cadence); silence means the worker is
+  // dead or its writes are failing, and the site is quietly serving the
+  // slow live fallback. Alert when stale > 5 min; the age windowing
+  // (first detection + hourly reminders) avoids a message every 5 min
+  // for the whole outage. Complements the worker's own write-failure
+  // pings, which cannot fire when the worker process itself is dead.
+  let matHeartbeatAgeSec: number | null = null;
+  if (storeConfigured()) {
+    try {
+      const hb = await readHeartbeat();
+      matHeartbeatAgeSec = hb ? Math.floor(Date.now() / 1000 - hb) : null;
+      const age = matHeartbeatAgeSec;
+      const stale = age === null || age > 300;
+      const firstDetection = age !== null && age <= 1200;
+      const hourlyReminder = age !== null && age % 3600 < 300;
+      if (webhook && stale && (age === null || firstDetection || hourlyReminder)) {
+        const ageTxt = age === null ? "no heartbeat key at all" : `${Math.round(age / 60)} min old`;
+        await fetch(webhook, {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            text: `🔴 OCB materialize worker heartbeat is stale (${ageTxt}) — pages are on the slow live fallback. Check the Railway service ocb-materialize-worker.`,
+          }),
+        }).catch((err) => console.error("slack heartbeat alert failed:", err));
+      }
+    } catch (err) {
+      console.error("materialize heartbeat check failed:", err);
+    }
+  }
+
   // Pre-warm the per-bench unstable_cache + KV snapshot layer. This
   // single call is what spares an unlucky cold-start user a draft render
   // when Prom is slow: it keeps each bench's runtime-cache entry fresh
@@ -233,6 +265,7 @@ export async function GET(req: NextRequest) {
     checked: liveSpecs.length,
     transitions: transitions.length,
     sent: sent.length,
+    matHeartbeatAgeSec,
     dryRun: !webhook,
     transitionsList: transitions,
     prewarm: { count: prewarmCount, err: prewarmErr },

src/lib/materialize/store.ts

@@ -94,6 +94,17 @@ export async function publishSnapshot(
   }
 }
 
+/** Refresh the current blob's safety-net TTL without rewriting it.
+ *  Called when the worker decides to KEEP the previous snapshot (bench
+ *  collapsed this sweep): without this, a bench that stays broken for
+ *  longer than BLOB_TTL_SEC would silently lose its carried-forward
+ *  data when the blob expires, defeating "the data is always there". */
+export async function touchSnapshot(slug: string, sig: string): Promise<void> {
+  const hash = await redis(["GET", matKeys.pointer(slug, sig)], 5_000);
+  if (typeof hash !== "string" || !hash) return;
+  await redis(["EXPIRE", matKeys.blob(slug, sig, hash), BLOB_TTL_SEC], 5_000);
+}
+
 export async function heartbeat(now = Date.now()): Promise<void> {
   await redis(["SET", matKeys.heartbeat, String(Math.floor(now / 1000))]);
 }

worker/index.ts

@@ -37,6 +37,7 @@ import {
   publishSnapshot,
   readMaterialized,
   storeConfigured,
+  touchSnapshot,
 } from "@/lib/materialize/store";
 import type { Benchmark, MetricPanel } from "@/types/benchmark";
 import type { Spec } from "@/lib/spec-schema";
@@ -160,6 +161,10 @@ async function materializeOne(
     // the previous snapshot simply ages (staleness surfaces it).
     if (prev) {
       console.warn(`[worker] ${spec.slug}/${sig || "all"} collapsed, keeping previous snapshot`);
+      // Keep the kept snapshot alive: refresh its safety-net TTL so a
+      // bench that stays broken does not lose its carried data when the
+      // blob would otherwise expire.
+      await touchSnapshot(spec.slug, sig).catch(() => {});
       return;
     }
     // No previous snapshot: publish the draft honestly (first boot during
@@ -209,6 +214,42 @@ async function inBatches<T>(items: T[], n: number, fn: (t: T) => Promise<void>)
   }
 }
 
+// ─── Self-alerting ───────────────────────────────────────────────────
+// The worker is the only writer; if its writes start failing (store
+// full, creds rotated, Upstash down) the site silently falls back to
+// the live path and NOBODY notices — that exact failure ran 10 hours
+// unnoticed on 2026-06-12 (quota blowout). Ping Slack after 3
+// consecutive failed heartbeats, then hourly reminders while broken.
+let writeFailStreak = 0;
+async function slackPing(text: string): Promise<void> {
+  const webhook = process.env.SLACK_WEBHOOK_URL?.trim();
+  if (!webhook) return;
+  await fetch(webhook, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ text }),
+    signal: AbortSignal.timeout(8000),
+  }).catch((e) => console.warn(`[worker] slack ping failed: ${e.message}`));
+}
+async function noteHeartbeat(ok: boolean, err?: unknown): Promise<void> {
+  if (ok) {
+    if (writeFailStreak >= 3) {
+      await slackPing("✅ OCB materialize worker: store writes recovered");
+    }
+    writeFailStreak = 0;
+    return;
+  }
+  writeFailStreak++;
+  const detail = err instanceof Error ? err.message : String(err);
+  console.warn(`[worker] heartbeat: ${detail}`);
+  // Fire on the 3rd consecutive failure, then every ~60 sweeps (~1h).
+  if (writeFailStreak === 3 || writeFailStreak % 60 === 0) {
+    await slackPing(
+      `🔴 OCB materialize worker: store writes failing for ${writeFailStreak} sweeps — site is serving the slow live fallback. Last error: ${detail.slice(0, 300)}`,
+    );
+  }
+}
+
 async function sweep(iteration: number): Promise<void> {
   const specs = await loadSpecsUncached();
   const t0 = Date.now();
@@ -221,7 +262,10 @@ async function sweep(iteration: number): Promise<void> {
     }
   });
   console.log(`[worker] tierA done in ${((Date.now() - t0) / 1000).toFixed(1)}s (${specs.length} benches)`);
-  await heartbeat().catch((e) => console.warn(`[worker] heartbeat: ${e.message}`));
+  await heartbeat().then(
+    () => noteHeartbeat(true),
+    (e) => noteHeartbeat(false, e),
+  );
 
   if (iteration % VARIANT_EVERY === 0) {
     const tB0 = Date.now();