Phase 1 aggregate broadcast (VPS-served): worker → static JSON + revalidate hook

src/app/api/internal/revalidate-aggregate/route.ts

@@ -0,0 +1,40 @@
+import { timingSafeEqual } from "node:crypto";
+import { revalidateTag } from "next/cache";
+import { NextResponse } from "next/server";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+/**
+ * Worker-driven CDN cache purge for the aggregate bench snapshot.
+ *
+ * The materialize worker POSTs here after successfully writing a fresh
+ * aggregate JSON to the VPS-served static path. This route invalidates
+ * the `bench-aggregate` cache tag so the next site request re-fetches
+ * the aggregate URL, instead of waiting for the ISR revalidate window.
+ *
+ * Auth: bearer token in `Authorization` header, `REVALIDATE_TOKEN` env,
+ * timing-safe compare. Fails closed when the env is missing.
+ */
+export async function POST(req: Request): Promise<NextResponse> {
+  const secret = (process.env.REVALIDATE_TOKEN ?? "").trim();
+  if (!secret) {
+    return NextResponse.json(
+      { error: "no_secret_configured" },
+      { status: 503 },
+    );
+  }
+  const header = (req.headers.get("authorization") ?? "").trim();
+  const expected = `Bearer ${secret}`;
+  const ok =
+    header.length === expected.length &&
+    timingSafeEqual(Buffer.from(header), Buffer.from(expected));
+  if (!ok) {
+    return NextResponse.json({ error: "unauthorized" }, { status: 401 });
+  }
+  // Next 16's revalidateTag takes a required cache-profile argument.
+  // "default" applies the standard purge semantics for the CDN + ISR
+  // cache layers this route is meant to invalidate.
+  revalidateTag("bench-aggregate", "default");
+  return NextResponse.json({ revalidated: true, tag: "bench-aggregate" });
+}

worker/index.ts

@@ -58,6 +58,7 @@ import { CHAINS } from "@/lib/chains";
 import { buildFeaturedLeadersFromStore } from "@/lib/search-featured";
 import type { Benchmark, MetricPanel } from "@/types/benchmark";
 import type { Spec } from "@/lib/spec-schema";
+import { publishAggregate } from "./publish-aggregate";
 
 const SWEEP_SEC = Number(process.env.SWEEP_SEC ?? 60);
 const VARIANT_EVERY = Number(process.env.VARIANT_EVERY ?? 5);
@@ -352,6 +353,26 @@ async function sweep(iteration: number): Promise<void> {
     (e) => noteHeartbeat(false, e),
   );
 
+  // Broadcast the fresh aggregate to a static file that Caddy serves
+  // publicly. The site then reads that URL from the CDN edge instead of
+  // fanning out ~150 concurrent Redis GETs per homepage render. See
+  // worker/publish-aggregate.ts for the full rationale. No-op when
+  // AGGREGATE_OUTPUT_PATH is unset (staged rollout).
+  if (process.env.AGGREGATE_OUTPUT_PATH) {
+    const pubStart = Date.now();
+    const result = await publishAggregate(specs);
+    const elapsedSec = ((Date.now() - pubStart) / 1000).toFixed(1);
+    if (result.ok) {
+      console.log(
+        `[worker] aggregate published in ${elapsedSec}s (${result.bytes} bytes, ${result.liveCount}/${result.total} live, revalidated=${result.revalidated})`,
+      );
+    } else {
+      console.warn(
+        `[worker] aggregate publish failed in ${elapsedSec}s: ${result.error}`,
+      );
+    }
+  }
+
   // Cohort snapshots used by the hub pages and the search dialog. Each
   // builder hits Prom directly (via the in-network http://ocb-prom:9090
   // URL), so they don't add load on the public reverse proxy. Failures

worker/publish-aggregate.ts

@@ -0,0 +1,138 @@
+/**
+ * Aggregate-snapshot broadcaster (self-hosted variant).
+ *
+ * After each Tier A sweep the worker calls `publishAggregate` to:
+ *   1. Read back every unfiltered bench snapshot it just wrote to Redis
+ *      (worker uses TCP direct via `OCB_REDIS_URL`, never SRH — so the
+ *      broadcast is insulated from the very failure mode this whole
+ *      pipeline exists to route around).
+ *   2. Assemble a lightweight envelope `{ v, builtAt, benches[] }`.
+ *   3. Write the JSON to `${AGGREGATE_OUTPUT_PATH}/latest.json` atomically
+ *      (write to `.tmp`, `rename` into place — readers never see a torn
+ *      snapshot). Caddy on the same VPS serves that path publicly.
+ *   4. Optionally POST the site's revalidate hook so the CDN cache tag
+ *      purges immediately, no waiting on the ~60 s revalidate window.
+ *
+ * Fail-soft: any step (missing config, disk full, network flap) degrades
+ * to a warning log. The worker's happy path is untouched, the site keeps
+ * reading via its existing Redis path via SRH. Once this proves stable,
+ * the site switches its aggregate reads to the public aggregate URL.
+ */
+
+import { mkdir, rename, writeFile } from "node:fs/promises";
+import path from "node:path";
+import type { Benchmark } from "@/types/benchmark";
+import type { Spec } from "@/lib/spec-schema";
+import { draftPlaceholderForSpec } from "@/lib/materialize/load";
+import { readMaterialized } from "@/lib/materialize/store";
+
+export type PublishResult = {
+  ok: boolean;
+  filePath?: string;
+  bytes?: number;
+  total?: number;
+  liveCount?: number;
+  draftCount?: number;
+  revalidated?: boolean;
+  error?: string;
+};
+
+const AGGREGATE_FILENAME = "latest.json";
+
+export async function publishAggregate(specs: Spec[]): Promise<PublishResult> {
+  const outputDir = process.env.AGGREGATE_OUTPUT_PATH;
+  if (!outputDir) {
+    return { ok: false, error: "AGGREGATE_OUTPUT_PATH not set" };
+  }
+
+  const benches: Benchmark[] = [];
+  let liveCount = 0;
+  let draftCount = 0;
+
+  for (const spec of specs) {
+    let bench: Benchmark | null = null;
+    try {
+      const snap = await readMaterialized(spec.slug, "");
+      bench = snap?.bench ?? null;
+    } catch (err) {
+      console.warn(
+        `[publish-aggregate] read ${spec.slug} failed: ${err instanceof Error ? err.message : err}`,
+      );
+    }
+    if (bench) {
+      benches.push(bench);
+      if (bench.status === "live") liveCount += 1;
+      else draftCount += 1;
+    } else {
+      benches.push(draftPlaceholderForSpec(spec));
+      draftCount += 1;
+    }
+  }
+
+  benches.sort((a, b) => (a.number ?? "").localeCompare(b.number ?? ""));
+
+  const envelope = {
+    v: 1,
+    builtAt: Date.now(),
+    total: benches.length,
+    liveCount,
+    draftCount,
+    benches,
+  };
+  const body = JSON.stringify(envelope);
+  const bytes = Buffer.byteLength(body);
+
+  const finalPath = path.join(outputDir, AGGREGATE_FILENAME);
+  const tmpPath = `${finalPath}.tmp`;
+
+  try {
+    await mkdir(outputDir, { recursive: true });
+    // Atomic write: readers (Caddy `file_server`) only ever open a
+    // fully-written file. `rename` on the same filesystem is atomic on
+    // POSIX so no torn reads possible.
+    await writeFile(tmpPath, body, "utf-8");
+    await rename(tmpPath, finalPath);
+  } catch (err) {
+    return {
+      ok: false,
+      error: `write failed: ${err instanceof Error ? err.message : String(err)}`,
+      bytes,
+      total: benches.length,
+      liveCount,
+      draftCount,
+    };
+  }
+
+  let revalidated = false;
+  const siteUrl = (process.env.SITE_URL ?? "").replace(/\/+$/, "");
+  const revalidateToken = process.env.REVALIDATE_TOKEN;
+  if (siteUrl && revalidateToken) {
+    try {
+      const res = await fetch(`${siteUrl}/api/internal/revalidate-aggregate`, {
+        method: "POST",
+        headers: { Authorization: `Bearer ${revalidateToken}` },
+        signal: AbortSignal.timeout(5_000),
+      });
+      revalidated = res.ok;
+      if (!res.ok) {
+        console.warn(
+          `[publish-aggregate] revalidate hook returned ${res.status} ${res.statusText}`,
+        );
+      }
+    } catch (err) {
+      console.warn(
+        `[publish-aggregate] revalidate hook failed: ${err instanceof Error ? err.message : err}`,
+      );
+    }
+  }
+
+  return {
+    ok: true,
+    filePath: finalPath,
+    bytes,
+    total: benches.length,
+    liveCount,
+    draftCount,
+    revalidated,
+  };
+}