From 00238e782bb372a614b6f56292cd1245c4349433 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Fri, 12 Jun 2026 12:09:49 +0200 Subject: [PATCH] feat(materialize): freshness alerting + keep-alive ttl for collapsed benches (step 7) --- src/app/api/cron/health-check/route.ts | 33 ++++++++++++++++++ src/lib/materialize/store.ts | 11 ++++++ worker/index.ts | 46 +++++++++++++++++++++++++- 3 files changed, 89 insertions(+), 1 deletion(-) diff --git a/src/app/api/cron/health-check/route.ts b/src/app/api/cron/health-check/route.ts index 7892bede..35575e66 100644 --- a/src/app/api/cron/health-check/route.ts +++ b/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 }, diff --git a/src/lib/materialize/store.ts b/src/lib/materialize/store.ts index 2ce3f3bc..7da6a242 100644 --- a/src/lib/materialize/store.ts +++ b/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 { + 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 { await redis(["SET", matKeys.heartbeat, String(Math.floor(now / 1000))]); } diff --git a/worker/index.ts b/worker/index.ts index 43176611..6392d9a4 100644 --- a/worker/index.ts +++ b/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(items: T[], n: number, fn: (t: T) => Promise) } } +// ─── 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 { + 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 { + 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 { const specs = await loadSpecsUncached(); const t0 = Date.now(); @@ -221,7 +262,10 @@ async function sweep(iteration: number): Promise { } }); 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();