From 0db6836e0fbaa94222bc830cd7e6f2989d26cdae Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Sat, 18 Jul 2026 19:25:41 +0200 Subject: [PATCH] Phase 1 aggregate broadcast (VPS-served): worker writes /state/aggregate/latest.json after tier A + revalidate hook (gated on AGGREGATE_OUTPUT_PATH) --- .../internal/revalidate-aggregate/route.ts | 40 +++++ worker/index.ts | 21 +++ worker/publish-aggregate.ts | 138 ++++++++++++++++++ 3 files changed, 199 insertions(+) create mode 100644 src/app/api/internal/revalidate-aggregate/route.ts create mode 100644 worker/publish-aggregate.ts diff --git a/src/app/api/internal/revalidate-aggregate/route.ts b/src/app/api/internal/revalidate-aggregate/route.ts new file mode 100644 index 00000000..4f2172ee --- /dev/null +++ b/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 { + 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" }); +} diff --git a/worker/index.ts b/worker/index.ts index ddac2b5f..9a9eedaf 100644 --- a/worker/index.ts +++ b/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 { (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 diff --git a/worker/publish-aggregate.ts b/worker/publish-aggregate.ts new file mode 100644 index 00000000..c7a2a627 --- /dev/null +++ b/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 { + 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, + }; +}