diff --git a/docs/build-monitor.md b/docs/build-monitor.md index 7dbc26a5..e1ac6afe 100644 --- a/docs/build-monitor.md +++ b/docs/build-monitor.md @@ -32,6 +32,32 @@ checks stop builds. The run selected on either tab carries over to the other. same run id, except that a collector record that stopped before the run ended never hides an end the Hugging Face copy recorded after the record's latest update. The Candidate tab applies the same rule to the runs it lists. + + Uploads to Hugging Face never pass through the collector, so its record + always says delivery mode `collector` with no counts. A collector record + that replaces a copy keeps the copy's upload record (files attempted and + uploaded, or a failed write check), and a run's uploads show as `unknown` + only when it has no copy that records them. + + Every Staging route reads the collector's run list (the country's 60 most + recent runs) through one cache on each server instance. On Vercel the + Candidate tab's selected-run route (`staging/run`, `maxDuration` 300) runs + in a different function from the Build progress routes and the Candidate + run list (120), because Vercel groups routes into one function only when + their settings match, so it keeps its own cache. An answer is + reused for 15 seconds, and a failed request is tried again on the next + call. The Candidate tab waits at most 5 seconds for the list. After that it + uses the list's latest earlier answer if that answered less than 60 seconds + ago (two of its 30-second polls), and otherwise shows the Hugging Face + states with a note; the late answer serves the next poll. A request still + in flight is shared only for 5 seconds after it was made, and a later call + makes its own request: once the Candidate tab has stopped waiting for a + request, nothing keeps it running, and Vercel can suspend the instance + with it in flight. Every collector request needs the collector's read + credential from Google Secret Manager, and that read follows the same + rule: a read still in flight is shared for 5 seconds after it began, and a + read that answered is reused for 60 seconds after it began. Each run's own + record is read on every request, as before. - **Local run folders.** The dashboard can still inspect legacy or diagnostic run folders on the same machine. Point it at the directory that holds them: @@ -85,6 +111,13 @@ The forecast compares the run with finished runs of the same pipeline from the same source (version 2 and 3 runs by pipeline id; version 1 US runs as one release pipeline), up to the 12 most recent. +Smoke runs (run kind `smoke`) and dry runs (`dry_run`) form pipelines of +their own, labelled "(smoke)" and "(dry run)", with their own history and +forecasts. A dry run checks its configuration, pins its inputs and stops, so +it would otherwise count as a passed run that ended after its first two +stages. The UK spine build registers its own pipeline id, `uk-frs-spine`, so +it is never compared with the calibration runs that read its output. + Runs of one pipeline can still differ in what they do: UK local candidates kept the pipeline id `uk-local-candidate` 0.1.0 when cloning and surface resolution moved into target compilation (September against October 2026). So a past run diff --git a/frontend/app/api/microcosm/staging/runs/route.ts b/frontend/app/api/microcosm/staging/runs/route.ts index fdabce64..66988b8c 100644 --- a/frontend/app/api/microcosm/staging/runs/route.ts +++ b/frontend/app/api/microcosm/staging/runs/route.ts @@ -13,7 +13,8 @@ import { scrub } from "@policyengine/calibration-tree/latest-artifact"; export const dynamic = "force-dynamic"; export const revalidate = 0; export const runtime = "nodejs"; -// The collector lookup can add two 30-second request timeouts. +// The collector lookup waits at most 5 seconds for the shared run list, but a +// run's own record can still take its 30-second request timeout. export const maxDuration = 120; export interface StagingRunsRouteDependencies { diff --git a/frontend/lib/microcosm/build-monitor.test.ts b/frontend/lib/microcosm/build-monitor.test.ts index 828c241c..1f3da503 100644 --- a/frontend/lib/microcosm/build-monitor.test.ts +++ b/frontend/lib/microcosm/build-monitor.test.ts @@ -9,6 +9,7 @@ import { type BuildTimeline, buildTimeline, calibrationRate, + comparableHistory, stopStatistics, failureReason, forecastCompletion, @@ -258,6 +259,84 @@ describe("buildTimeline, version 2 lifecycle", () => { expect(smoke.pipeline_label).toBe("UK FRS calibration (smoke)"); }); + // A calibration dry run (`--dry-run`) registers run kind `dry_run`: it + // checks its configuration, pins its inputs and stops, so its two stages + // are the start of every full run's and it ends in minutes. + function dryRun(runId: string): BuildRunDocuments { + return { + run_id: runId, + source: "staging", + country: "uk", + progress: { + schema_version: 2, + run_id: runId, + country_code: "GB", + pipeline: { id: "uk-frs-calibration", version: "collector-v1" }, + run_kind: "dry_run", + started_at: at(-min(20)), + updated_at: at(-min(2)), + status: "completed", + current_stage: "complete", + }, + run_manifest: null, + calibration_progress: null, + events: [ + v2Event(1, -min(20), "created", "started"), + v2Event(2, -min(20), "preflight", "started"), + v2Event(3, -min(19), "preflight", "completed"), + v2Event(4, -min(19), "input_pinning", "started"), + v2Event(5, -min(2), "input_pinning", "completed"), + v2Event(6, -min(2), "complete", "completed"), + ], + }; + } + + const calibrationRun: BuildRunDocuments = { + ...documents, + progress: { ...documents.progress!, run_kind: "calibration" }, + }; + + test("keeps dry runs apart from full runs of the same pipeline", () => { + const dry = buildTimeline(dryRun("uk-dry"), T0 + min(9)); + + expect(dry.state).toBe("passed"); + expect(dry.pipeline).toBe("GB:uk-frs-calibration:dry_run"); + expect(dry.pipeline_label).toBe("UK FRS calibration (dry run)"); + }); + + test("a dry run is not in a calibration run's history or forecast", () => { + const live = buildTimeline(calibrationRun, T0 + min(9)); + const dry = buildTimeline(dryRun("uk-dry"), T0 + min(9)); + + expect(live.pipeline).toBe("GB:uk-frs-calibration"); + expect(comparableHistory(live, [live, dry])).toEqual([]); + expect(forecastCompletion(live, [live, dry], T0 + min(9))?.basis_runs).toEqual([]); + }); + + test("a spine build is a pipeline of its own", () => { + // The spine build registers pipeline `uk-frs-spine` and run kind `spine`, + // so it never shares a calibration run's history. + const spine = buildTimeline( + { + ...calibrationRun, + run_id: "uk-spine", + progress: { + ...calibrationRun.progress!, + run_id: "uk-spine", + pipeline: { id: "uk-frs-spine", version: "collector-v1" }, + run_kind: "spine", + status: "completed", + }, + }, + T0 + min(9), + ); + const live = buildTimeline(calibrationRun, T0 + min(9)); + + expect(spine.pipeline).toBe("GB:uk-frs-spine"); + expect(spine.pipeline_label).toBe("UK FRS spine"); + expect(comparableHistory(live, [live, spine])).toEqual([]); + }); + test("reads the solver rate from calibration progress", () => { const run = buildTimeline(documents, T0 + min(9)); const rate = calibrationRate(run.calibration)!; diff --git a/frontend/lib/microcosm/build-monitor.ts b/frontend/lib/microcosm/build-monitor.ts index 73e15628..5061ad21 100644 --- a/frontend/lib/microcosm/build-monitor.ts +++ b/frontend/lib/microcosm/build-monitor.ts @@ -96,6 +96,9 @@ export interface StageWork { // Whether a run's telemetry reached the staging repository. `local_only` // runs kept it on the build machine (no write token, uploads failed, or a // local-only run); `uploaded` says how many files reached the repository. +// The collector does not know (its record says delivery mode "collector"), +// so a collector record reports the uploads its run's Hugging Face copy +// records, and `unknown` when there is no such record. export interface RunDelivery { uploads: "local_only" | "uploaded" | "disabled" | "unknown"; repository: string | null; @@ -257,6 +260,16 @@ export function formatStageName(stage: string): string { return text ? text[0].toUpperCase() + text.slice(1) : stage; } +// Run kinds that are not comparable with full runs of the same pipeline, with +// the label suffix for each. A smoke run builds a synthetic fixture; a dry run +// (`--dry-run`) checks its configuration, pins its inputs and stops. The run +// kind is part of their pipeline key, so they have their own group, history +// and forecasts. +const SEPARATE_RUN_KINDS: ReadonlyMap = new Map([ + ["smoke", "smoke"], + ["dry_run", "dry run"], +]); + function pipelineIdentity( documents: BuildRunDocuments, schemaVersion: 1 | 2 | 3, @@ -266,13 +279,11 @@ function pipelineIdentity( const pipelineId = str(pipeline?.id); const country = str(identity?.country_code) ?? documents.country.toUpperCase(); if (schemaVersion >= 2 && pipelineId) { - // A smoke run on a synthetic fixture is not comparable with a full run of - // the same pipeline, so the run kind is part of the key. const kind = str(identity?.run_kind); - const smoke = kind === "smoke"; + const kindLabel = kind == null ? undefined : SEPARATE_RUN_KINDS.get(kind); return { - key: `${country}:${pipelineId}${smoke ? ":smoke" : ""}`, - label: `${formatStageName(pipelineId)}${smoke ? " (smoke)" : ""}`, + key: `${country}:${pipelineId}${kindLabel ? `:${kind}` : ""}`, + label: `${formatStageName(pipelineId)}${kindLabel ? ` (${kindLabel})` : ""}`, }; } return { @@ -396,6 +407,52 @@ function runDelivery(documents: BuildRunDocuments): RunDelivery | null { }; } +function uploadsKnown(delivery: RunDelivery | null): boolean { + return delivery != null && delivery.uploads !== "unknown"; +} + +// A collector record with its run's upload record taken from the run's Hugging +// Face copy: the copy's failed write check (version 1) or delivery block +// (version 2 and later). Uploads to Hugging Face never pass through the +// collector, so its own delivery block always says mode "collector" with no +// counts. The record is unchanged when the copy records no uploads. +export function withHuggingFaceUploads( + collector: BuildRunDocuments, + huggingFace: BuildRunDocuments, +): BuildRunDocuments { + const check = obj(huggingFace.run_manifest?.delivery_check); + const delivery = + obj(huggingFace.progress?.delivery) ?? obj(huggingFace.run_manifest?.delivery); + if (!check && !delivery) return collector; + let progress = collector.progress; + if (progress && "delivery" in progress) { + const { delivery: _collectorDelivery, ...rest } = progress; + progress = rest; + } + return { + ...collector, + progress, + run_manifest: { + ...collector.run_manifest, + ...(check ? { delivery_check: check } : {}), + ...(delivery ? { delivery } : {}), + }, + }; +} + +// A run read again from the collector, keeping the uploads `listed` (the same +// run in the monitor's list, where they came from the Hugging Face copy) when +// the collector's record does not know them. +export function keepListedUploads( + fresh: BuildTimeline, + listed: BuildTimeline | null, +): BuildTimeline { + if (uploadsKnown(fresh.delivery) || !listed || !uploadsKnown(listed.delivery)) { + return fresh; + } + return { ...fresh, delivery: listed.delivery }; +} + function runIdentity(manifest: JsonObject | null): RunIdentity | null { const identity = obj(manifest?.identity); if (!identity) return null; diff --git a/frontend/lib/microcosm/build-runs.test.ts b/frontend/lib/microcosm/build-runs.test.ts index 1f505054..5924b5af 100644 --- a/frontend/lib/microcosm/build-runs.test.ts +++ b/frontend/lib/microcosm/build-runs.test.ts @@ -1,15 +1,23 @@ -import { describe, expect, spyOn, test } from "bun:test"; +import { beforeEach, describe, expect, jest, spyOn, test } from "bun:test"; import { type BuildRunDocuments, + type RunDelivery, buildTimeline, } from "@/lib/microcosm/build-monitor"; import { type HostedRunLoaders, + collectorRunLists, loadBuildRuns, loadCollectorTimelines, + mergeHostedDocuments, refreshHostedRun, } from "@/lib/microcosm/build-runs"; +import { + UK_STAGED_RUN_ID, + ukStagedCollectorDocuments, + ukStagedHuggingFaceDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; import { FAKE_COLLECTOR_URL, fakeTelemetryCollector, @@ -18,6 +26,10 @@ import { secretManager } from "@/lib/server/secret-manager"; const NOW = Date.parse("2026-10-05T12:00:00Z"); +// Each test's fake collector answers for itself, not from an earlier test's +// cached run list. +beforeEach(() => collectorRunLists.clear()); + function documents( runId: string, status: "running" | "completed", @@ -383,3 +395,322 @@ describe("loadCollectorTimelines for the Candidate tab", () => { expect(loaded.note).toContain("500"); }); }); + +describe("uploads to Hugging Face for runs the collector records", () => { + // The collector never learns whether a run's files reached Hugging Face, so + // the monitor reads the uploads from the run's Hugging Face copy. + const STAGED_NOW_MS = Date.parse("2026-10-22T17:11:00Z"); + const UPLOADED: RunDelivery = { + uploads: "uploaded", + repository: "policyengine/populace-uk-staging", + reason: null, + attempts: 39, + successes: 39, + }; + + function listedRun(huggingFace: BuildRunDocuments[]) { + return mergeHostedDocuments([ukStagedCollectorDocuments()], huggingFace, STAGED_NOW_MS) + .map((documents) => buildTimeline(documents, STAGED_NOW_MS)) + .find((timeline) => timeline.run_id === UK_STAGED_RUN_ID)!; + } + + test("the list shows the uploads the Hugging Face copy records", () => { + const run = listedRun([ukStagedHuggingFaceDocuments()]); + + // The collector's record still sets the state and the end. + expect(run.state).toBe("passed"); + expect(run.ended_ms).toBe(Date.parse("2026-10-21T20:36:30Z")); + expect(run.delivery).toEqual(UPLOADED); + }); + + test("uploads stay unknown when the run has no Hugging Face copy", () => { + const run = listedRun([]); + + expect(run.delivery?.uploads).toBe("unknown"); + }); + + test("uploads stay unknown when the Hugging Face copy records none", () => { + const copy = ukStagedHuggingFaceDocuments(); + const { delivery: _progressDelivery, ...progress } = copy.progress!; + const run = listedRun([{ ...copy, progress, run_manifest: null }]); + + expect(run.delivery?.uploads).toBe("unknown"); + }); + + test("refreshing the selected run from the collector keeps the listed uploads", async () => { + const listed = listedRun([ukStagedHuggingFaceDocuments()]); + let historyCalls = 0; + + const run = await refreshHostedRun( + listed, + "uk", + UK_STAGED_RUN_ID, + STAGED_NOW_MS, + loaders({ + loadCollector: async () => ukStagedCollectorDocuments(), + loadHistory: async () => { + historyCalls += 1; + return ukStagedHuggingFaceDocuments(); + }, + }), + ); + + expect(run?.state).toBe("passed"); + expect(run?.delivery).toEqual(UPLOADED); + expect(historyCalls).toBe(0); + }); +}); + +describe("one collector run list for every Staging route", () => { + test("the Build progress list and both Candidate routes share one list request", async () => { + // The be country has no staging repository, so Build progress reads only + // the collector. + const collector = fakeTelemetryCollector({ countryCode: "BE", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + + await withCollector(collector.fetch, async () => { + await loadBuildRuns("staging", "be"); + // The Candidate tab's run list and its selected run. + await loadCollectorTimelines("be", [runningRunId], NOW); + await loadCollectorTimelines("be", [runningRunId], NOW); + }); + + expect(collector.listRequests).toHaveLength(1); + // A running run's own record is still read for every request. + expect( + collector.documentRequests.filter((runId) => runId === runningRunId), + ).toHaveLength(3); + }); + + test("the list is read again once it is 15 seconds old", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + const listRequestsAt: number[] = []; + + jest.useFakeTimers(); + try { + await withCollector(collector.fetch, async () => { + await loadCollectorTimelines("am", [runningRunId], NOW); + listRequestsAt.push(collector.listRequests.length); + jest.advanceTimersByTime(14_000); + await loadCollectorTimelines("am", [runningRunId], NOW); + listRequestsAt.push(collector.listRequests.length); + jest.advanceTimersByTime(2_000); + await loadCollectorTimelines("am", [runningRunId], NOW); + listRequestsAt.push(collector.listRequests.length); + }); + } finally { + jest.useRealTimers(); + } + + expect(listRequestsAt).toEqual([1, 1, 2]); + }); + + test("a failed list request is not reused", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + let listRequestCount = 0; + const failingOnce = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs" && listRequestCount++ === 0) { + return Response.json({ detail: "Unavailable." }, { status: 503 }); + } + return collector.fetch(input); + }) as typeof fetch; + + const [failed, answered] = await withCollector(failingOnce, async () => [ + await loadCollectorTimelines("am", [runningRunId], NOW), + await loadCollectorTimelines("am", [runningRunId], NOW), + ]); + + expect(failed.note).toContain("503"); + expect(answered.note).toBeNull(); + expect(answered.timelines.get(runningRunId)?.state).toBe("running"); + }); + + test("the Candidate tab waits at most 5 seconds for the list, and a late answer serves the next poll", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + let answerList!: () => void; + const listAnswer = new Promise((resolve) => { + answerList = resolve; + }); + const slowList = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs") await listAnswer; + return collector.fetch(input); + }) as typeof fetch; + + jest.useFakeTimers(); + let firstPoll: Awaited>; + let nextPoll: Awaited>; + try { + [firstPoll, nextPoll] = await withCollector(slowList, async () => { + const pending = loadCollectorTimelines("am", [runningRunId], NOW); + // The first poll's list request, which carries on after the wait. + const firstList = collectorRunLists.recent("am"); + jest.advanceTimersByTime(5_000); + // The collector answers only after the wait is over. + answerList(); + const first = await pending; + await firstList; + return [first, await loadCollectorTimelines("am", [runningRunId], NOW)]; + }); + } finally { + jest.useRealTimers(); + } + + expect(firstPoll.timelines.size).toBe(0); + expect(firstPoll.note).toContain("did not answer within 5 seconds"); + expect(firstPoll.note).toContain("Hugging Face"); + expect(nextPoll.note).toBeNull(); + expect(nextPoll.timelines.get(runningRunId)?.state).toBe("running"); + expect(collector.listRequests).toHaveLength(1); + }); + + test("a late answer serves the next 30-second poll while the collector stays slow", async () => { + // The Candidate tab polls every 30 seconds. A list that answers 6 seconds + // after each request is more than 15 seconds old at the next poll, whose + // own request is again slower than the 5-second wait. + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + const unansweredListRequests: Array<() => void> = []; + const slowList = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs") { + await new Promise((resolve) => unansweredListRequests.push(resolve)); + } + return collector.fetch(input); + }) as typeof fetch; + + jest.useFakeTimers(); + let firstPoll: Awaited>; + let nextPoll: Awaited>; + try { + [firstPoll, nextPoll] = await withCollector(slowList, async () => { + const firstPending = loadCollectorTimelines("am", [runningRunId], NOW); + // The first poll's list request, which carries on after the wait. + const firstList = collectorRunLists.recent("am"); + jest.advanceTimersByTime(5_000); + const first = await firstPending; + // The first request answers 6 seconds after it was made. + jest.advanceTimersByTime(1_000); + unansweredListRequests.shift()?.(); + await firstList; + // The next poll, 30 seconds after the first. + jest.advanceTimersByTime(24_000); + const nextPending = loadCollectorTimelines("am", [runningRunId], NOW); + jest.advanceTimersByTime(5_000); + const next = await nextPending; + for (const answer of unansweredListRequests.splice(0)) answer(); + return [first, next]; + }); + } finally { + jest.useRealTimers(); + } + + expect(firstPoll.note).toContain("did not answer within 5 seconds"); + expect(nextPoll.note).toBeNull(); + expect(nextPoll.timelines.get(runningRunId)?.state).toBe("running"); + }); + + test("an answer a minute old no longer stands in for a slow list", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + let listRequestCount = 0; + const unansweredListRequests: Array<() => void> = []; + // The first list request answers at once; later ones are slow. + const slowAfterFirst = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs" && listRequestCount++ > 0) { + await new Promise((resolve) => unansweredListRequests.push(resolve)); + } + return collector.fetch(input); + }) as typeof fetch; + + jest.useFakeTimers(); + let latePoll: Awaited>; + try { + latePoll = await withCollector(slowAfterFirst, async () => { + await loadCollectorTimelines("am", [runningRunId], NOW); + jest.advanceTimersByTime(60_000); + const latePending = loadCollectorTimelines("am", [runningRunId], NOW); + jest.advanceTimersByTime(5_000); + const late = await latePending; + for (const answer of unansweredListRequests.splice(0)) answer(); + return late; + }); + } finally { + jest.useRealTimers(); + } + + expect(latePoll.timelines.size).toBe(0); + expect(latePoll.note).toContain("did not answer within 5 seconds"); + }); + + test("a call made while the list request is in flight shares it", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + let answerList!: () => void; + const listAnswer = new Promise((resolve) => { + answerList = resolve; + }); + const slowList = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs") await listAnswer; + return collector.fetch(input); + }) as typeof fetch; + + jest.useFakeTimers(); + let polls: Array>>; + try { + polls = await withCollector(slowList, async () => { + const firstPending = loadCollectorTimelines("am", [runningRunId], NOW); + // A second call 4 seconds later, before the first one's wait is over. + jest.advanceTimersByTime(4_000); + const secondPending = loadCollectorTimelines("am", [runningRunId], NOW); + answerList(); + return Promise.all([firstPending, secondPending]); + }); + } finally { + jest.useRealTimers(); + } + + expect(polls.map((poll) => poll.note)).toEqual([null, null]); + expect(collector.listRequests).toHaveLength(1); + }); + + test("a list request left running after the wait is not shared by the next poll", async () => { + // Once every caller has stopped waiting, nothing keeps the request + // running: Vercel can suspend the instance with it in flight, and its + // overdue 30-second timeout fires when the next poll resumes the instance. + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + const runningRunId = collector.runs[0].run_id; + let listRequestCount = 0; + let failFirstList!: (error: Error) => void; + const firstListTimesOut = (async (input: RequestInfo | URL) => { + if (new URL(String(input)).pathname === "/v1/runs" && listRequestCount++ === 0) { + await new Promise((_resolve, reject) => { + failFirstList = reject; + }); + } + return collector.fetch(input); + }) as typeof fetch; + + jest.useFakeTimers(); + let nextPoll: Awaited>; + try { + nextPoll = await withCollector(firstListTimesOut, async () => { + const firstPending = loadCollectorTimelines("am", [runningRunId], NOW); + jest.advanceTimersByTime(5_000); + await firstPending; + // The next poll, 30 seconds after the first, finds the first request + // still in flight, and that request then fails with its timeout. + jest.advanceTimersByTime(25_000); + const nextPending = loadCollectorTimelines("am", [runningRunId], NOW); + failFirstList(new Error("The operation was aborted due to timeout")); + return nextPending; + }); + } finally { + jest.useRealTimers(); + } + + expect(nextPoll.note).toBeNull(); + expect(nextPoll.timelines.get(runningRunId)?.state).toBe("running"); + }); +}); diff --git a/frontend/lib/microcosm/build-runs.ts b/frontend/lib/microcosm/build-runs.ts index 9a62e2de..20f18e7c 100644 --- a/frontend/lib/microcosm/build-runs.ts +++ b/frontend/lib/microcosm/build-runs.ts @@ -9,6 +9,7 @@ import { buildTimeline, collectorMissedTheEnd, compactTimeline, + keepListedUploads, type StopStat, stopStatistics, forecastCompletion, @@ -16,6 +17,7 @@ import { phaseTotals, sameStageSequence, stageStatistics, + withHuggingFaceUploads, } from "@/lib/microcosm/build-monitor"; import { type CatalogGate, @@ -48,6 +50,23 @@ import { const STAGING_LIST_TTL_MS = 30_000; const STAGING_RUN_LIMIT = 60; +// How long one answer to the collector's run list serves the Staging routes +// on a server instance. The Candidate tab's two routes poll together every 30 +// seconds, as does the Build progress list. On Vercel, `staging/run` (whose +// `maxDuration` is 300) runs in a different function from `builds`, +// `builds/run` and `staging/runs` (120), so it keeps its own cache, and a poll +// makes one list request per country in each of the two functions. Each run's +// own record is read as before. +const COLLECTOR_LIST_TTL_MS = 15_000; +// How long the Candidate tab waits for the collector's run list. Past it, the +// tab uses the list's latest earlier answer, or shows the Hugging Face states +// when there is none. The request carries on while the server instance runs, +// and its answer serves the next poll, but a later call does not share it. +const COLLECTOR_LIST_WAIT_MS = 5_000; +// How old that earlier answer may be: two of the Candidate tab's 30-second +// polls. A list that answers after the wait is more than 15 seconds old at the +// next poll, and that poll's own request may be as slow. +const EARLIER_COLLECTOR_LIST_MAX_AGE_MS = 60_000; export interface BuildRunProblem { run_id: string; @@ -96,6 +115,111 @@ const stagingListCache = new Map< { expiresAt: number; promise: Promise } >(); +// The country's 60 most recent collector runs, shared by every caller: a list +// that answered is reused for `COLLECTOR_LIST_TTL_MS` after it answered, a +// request in flight is shared for `COLLECTOR_LIST_WAIT_MS` after it was made, +// and a failed request is dropped so the next caller asks again. +// +// An older request in flight may have no caller left: the Candidate tab stops +// waiting after `COLLECTOR_LIST_WAIT_MS`, and nothing keeps the request running +// after the response, so Vercel can suspend the instance with it in flight. A +// later caller that shared it would receive its overdue timeout error, so a +// later caller makes a new request instead. +const collectorLists = new Map< + MicrocosmCountry, + { + runs: Promise; + requestedAt: number; + answeredAt: number | null; + } +>(); +// The list of the latest request that answered for each country, kept while +// a newer request is in flight. +const answeredCollectorLists = new Map< + MicrocosmCountry, + { runs: CollectorRunSummary[]; requestedAt: number; answeredAt: number } +>(); +// Counts `clear()` calls, so that a request made before one keeps no answer. +let collectorListsGeneration = 0; + +export const collectorRunLists = { + recent(country: MicrocosmCountry): Promise { + const now = Date.now(); + const cached = collectorLists.get(country); + if ( + cached && + (cached.answeredAt == null + ? now - cached.requestedAt < COLLECTOR_LIST_WAIT_MS + : now - cached.answeredAt < COLLECTOR_LIST_TTL_MS) + ) { + return cached.runs; + } + const generation = collectorListsGeneration; + const entry = { + runs: loadCollectorRuns(country, STAGING_RUN_LIMIT), + requestedAt: now, + answeredAt: null as number | null, + }; + collectorLists.set(country, entry); + entry.runs.then( + (runs) => { + entry.answeredAt = Date.now(); + // An answer to a request a newer one replaced still counts, unless + // `clear()` ran since or a newer request has already answered. + const latest = answeredCollectorLists.get(country); + if ( + generation === collectorListsGeneration && + (!latest || latest.requestedAt <= entry.requestedAt) + ) { + answeredCollectorLists.set(country, { + runs, + requestedAt: entry.requestedAt, + answeredAt: entry.answeredAt, + }); + } + }, + () => { + if (collectorLists.get(country) === entry) collectorLists.delete(country); + }, + ); + return entry.runs; + }, + // The latest list that answered less than `maxAgeMs` ago, or null. + answeredWithin( + country: MicrocosmCountry, + maxAgeMs: number, + ): CollectorRunSummary[] | null { + const answered = answeredCollectorLists.get(country); + return answered && Date.now() - answered.answeredAt < maxAgeMs + ? answered.runs + : null; + }, + clear(): void { + collectorListsGeneration += 1; + collectorLists.clear(); + answeredCollectorLists.clear(); + }, +}; + +const WAIT_OVER = Symbol("wait over"); + +// `promise`'s value, or `WAIT_OVER` when it has not settled within `waitMs`. +// A rejection within the wait is thrown. +async function settledWithin( + promise: Promise, + waitMs: number, +): Promise { + let timer: ReturnType | undefined; + const waitOver = new Promise((resolve) => { + timer = setTimeout(() => resolve(WAIT_OVER), waitMs); + }); + try { + return await Promise.race([promise, waitOver]); + } finally { + clearTimeout(timer); + } +} + interface LoadedRuns { available: boolean; detail: string | null; @@ -253,7 +377,7 @@ async function loadCollectorDocuments( } async function loadCollector(country: MicrocosmCountry): Promise { - const summaries = await loadCollectorRuns(country, STAGING_RUN_LIMIT); + const summaries = await collectorRunLists.recent(country); const documents: BuildRunDocuments[] = []; const problems: BuildRunProblem[] = []; await Promise.all( @@ -282,12 +406,14 @@ const HUGGING_FACE_FALLBACK = // The collector's state for runs the Candidate tab lists from Hugging Face. // It reads the same records the Build progress tab does (the country's 60 -// most recent collector runs) and derives each state with the same -// `buildTimeline`, so the two tabs agree on every run the collector knows. +// most recent collector runs, from the shared list) and derives each state +// with the same `buildTimeline`, so the two tabs agree on every run the +// collector knows. It waits at most `waitMs` for the list. export async function loadCollectorTimelines( country: MicrocosmCountry, runIds: readonly string[], nowMs: number = Date.now(), + waitMs: number = COLLECTOR_LIST_WAIT_MS, ): Promise { const timelines = new Map(); if (!runIds.length) return { timelines, note: null }; @@ -301,7 +427,24 @@ export async function loadCollectorTimelines( note: `The telemetry collector is not configured on this server, so ${HUGGING_FACE_FALLBACK}.`, }; } - summaries = await loadCollectorRuns(country, STAGING_RUN_LIMIT); + const listed = await settledWithin(collectorRunLists.recent(country), waitMs); + if (listed !== WAIT_OVER) { + summaries = listed; + } else { + // The request carries on, and its answer serves a later poll. Until it + // answers, an earlier answer that is recent enough stands in. + const earlier = collectorRunLists.answeredWithin( + country, + EARLIER_COLLECTOR_LIST_MAX_AGE_MS, + ); + if (!earlier) { + return { + timelines, + note: `The telemetry collector did not answer within ${waitMs / 1000} seconds, so ${HUGGING_FACE_FALLBACK}.`, + }; + } + summaries = earlier; + } } catch (error) { return { timelines, @@ -337,7 +480,9 @@ export async function loadCollectorTimelines( // The hosted documents the build monitor shows, one per run id. During the // migration period a collector record replaces the run's historical Hugging // Face copy, unless the record stopped before an end the copy recorded later -// (`collectorMissedTheEnd`, the rule the Candidate tab uses too). +// (`collectorMissedTheEnd`, the rule the Candidate tab uses too). A record +// that replaces a copy keeps the copy's upload record, which the collector +// cannot know. export function mergeHostedDocuments( collector: readonly BuildRunDocuments[], huggingFace: readonly BuildRunDocuments[], @@ -356,7 +501,10 @@ export function mergeHostedDocuments( ) { continue; } - byId.set(document.run_id, document); + byId.set( + document.run_id, + copy ? withHuggingFaceUploads(document, copy) : document, + ); } return [...byId.values()]; } @@ -453,10 +601,11 @@ export async function refreshHostedRun( if (collectorRun) { // The list kept the Hugging Face copy when the collector's record // stopped before an end the copy recorded later; a fresh read of that - // record must not hide the end either. + // record must not hide the end either. Otherwise the fresh record wins, + // with the uploads the list read from the copy. return current && collectorMissedTheEnd(collectorRun, current) ? current - : collectorRun; + : keepListedUploads(collectorRun, current); } } if (!loaders.hasStagingHistory(country)) { diff --git a/frontend/lib/microcosm/test-support/collector-run-fixtures.ts b/frontend/lib/microcosm/test-support/collector-run-fixtures.ts index 99c92f24..c49db261 100644 --- a/frontend/lib/microcosm/test-support/collector-run-fixtures.ts +++ b/frontend/lib/microcosm/test-support/collector-run-fixtures.ts @@ -31,6 +31,12 @@ export const UK_BLOCKED_RUN_ID = export const UK_COMPLETED_HUGGING_FACE_RUN_ID = "uk-local-candidate-f100-s42-20261007T120029Z-2ee47c99"; +// A UK national calibration attempt that uploaded its telemetry (39 of 39 +// files) and staged its dataset. The collector and the Hugging Face copy both +// record it passing. +export const UK_STAGED_RUN_ID = + "uk-frs-calibration-attempt-20261021T195620Z-a858c100"; + export const US_RELEASE_CHECK_FAILURE = "Release gates failed: Critical fiscal target 'ssa_supplement.cy2024.oasdi_ssi_payments.social_security_benefits.payment_amount@2024' (Social Security benefits) has relative_error=-0.0641635, exceeding 0.05"; @@ -247,3 +253,97 @@ export function ukBlockedCollectorDocuments(): BuildRunDocuments { ], ); } + +// The collector's documents for the staged UK attempt. Its run manifest +// records delivery the way the collector always does: mode "collector" with no +// upload counts, because uploads to Hugging Face never pass through it. +export function ukStagedCollectorDocuments(): BuildRunDocuments { + const documents = collectorDocuments( + UK_STAGED_RUN_ID, + "uk", + { + country_code: "GB", + pipeline: { id: "uk-frs-calibration", version: "collector-v1" }, + run_kind: "calibration", + started_at: "2026-10-21T19:56:21Z", + updated_at: "2026-10-21T20:36:30Z", + heartbeat_at: "2026-10-21T20:36:02Z", + status: "completed", + current_stage: "complete", + }, + [ + { stage: "created", status: "started", time: "2026-10-21T19:56:21Z" }, + { stage: "preflight", status: "started", time: "2026-10-21T19:56:21Z" }, + { stage: "preflight", status: "completed", time: "2026-10-21T19:56:40Z" }, + { stage: "calibration", status: "started", time: "2026-10-21T20:05:00Z" }, + { stage: "calibration", status: "completed", time: "2026-10-21T20:25:00Z" }, + { stage: "dataset_staging", status: "started", time: "2026-10-21T20:33:00Z" }, + { stage: "dataset_staging", status: "completed", time: "2026-10-21T20:35:31Z" }, + { stage: "complete", status: "completed", time: "2026-10-21T20:36:30Z" }, + ], + ); + return { + ...documents, + run_manifest: { + schema_version: 2, + run_id: UK_STAGED_RUN_ID, + country_code: "GB", + status: "completed", + delivery: { mode: "collector", upload_attempts: null, upload_successes: null }, + }, + }; +} + +// The staged UK attempt's Hugging Face copy: the progress document its final +// upload wrote, with the writer's own count of uploaded files. +export function ukStagedHuggingFaceDocuments(): BuildRunDocuments { + const delivery = { + configured_repository: "policyengine/populace-uk-staging", + contract_version: 2, + enabled: true, + last_error_code: null, + mode: "local_and_remote", + opt_out_reason: null, + read_back: "not_requested", + run_id: UK_STAGED_RUN_ID, + upload_attempts: 39, + upload_successes: 39, + }; + return { + run_id: UK_STAGED_RUN_ID, + source: "staging", + country: "uk", + progress: { + schema_version: 3, + run_id: UK_STAGED_RUN_ID, + country_code: "GB", + pipeline: { id: "uk-frs-calibration", version: "0.1.0" }, + run_kind: "calibration", + started_at: "2026-10-21T19:56:21+00:00", + updated_at: "2026-10-21T20:36:24+00:00", + status: "completed", + current_stage: "complete", + delivery, + }, + run_manifest: { schema_version: 3, run_id: UK_STAGED_RUN_ID, status: "completed", delivery }, + calibration_progress: null, + events: [ + { + schema_version: 3, + sequence: 1, + event_type: "stage", + stage_id: "created", + status: "started", + timestamp: "2026-10-21T19:56:21+00:00", + }, + { + schema_version: 3, + sequence: 2, + event_type: "stage", + stage_id: "complete", + status: "completed", + timestamp: "2026-10-21T20:36:24+00:00", + }, + ], + }; +} diff --git a/frontend/lib/server/secret-manager.test.ts b/frontend/lib/server/secret-manager.test.ts index 055043bd..126b1374 100644 --- a/frontend/lib/server/secret-manager.test.ts +++ b/frontend/lib/server/secret-manager.test.ts @@ -25,6 +25,45 @@ test("shares an in-flight read and refreshes the cached secret", async () => { expect(access).toHaveBeenCalledTimes(2); }); +test("a read still in flight 5 seconds after it began is not shared", async () => { + // The build monitor's Candidate tab stops waiting for the collector's run + // list after 5 seconds, which can leave this read in flight with nothing + // keeping it running. Vercel can then suspend the instance, and the read's + // overdue 10-second timeout fires when a later call resumes the instance. + let now = 0; + let failFirstRead!: (error: Error) => void; + const access = mock(async () => encoded("fresh")); + access.mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + failFirstRead = reject; + }), + ); + const reader = createSecretReader(access, () => now); + const firstRead = reader.readVersion(resource); + now = 4_999; + const sharedRead = reader.readVersion(resource); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(access).toHaveBeenCalledTimes(1); + + now = 5_000; + const laterRead = reader.readVersion(resource); + failFirstRead(new Error("The operation was aborted due to timeout")); + const [first, shared, later] = await Promise.allSettled([ + firstRead, + sharedRead, + laterRead, + ]); + expect([first.status, shared.status]).toEqual(["rejected", "rejected"]); + expect(later).toEqual({ status: "fulfilled", value: "fresh" }); + expect(access).toHaveBeenCalledTimes(2); + + // A read that answered is still reused for 60 seconds after it began. + now = 64_999; + expect(await reader.readVersion(resource)).toBe("fresh"); + expect(access).toHaveBeenCalledTimes(2); +}); + test("keeps distinct secret versions in separate cache entries", async () => { const access = mock(async (name: string) => encoded(name.endsWith("/1") ? "first" : "latest"), diff --git a/frontend/lib/server/secret-manager.ts b/frontend/lib/server/secret-manager.ts index 91e94dc6..d348a05f 100644 --- a/frontend/lib/server/secret-manager.ts +++ b/frontend/lib/server/secret-manager.ts @@ -4,7 +4,16 @@ import { getVercelOidcToken } from "@vercel/oidc"; import { GoogleAuth, IdentityPoolClient } from "google-auth-library"; const CLOUD_PLATFORM_SCOPE = "https://www.googleapis.com/auth/cloud-platform"; +// How long a read that answered is reused, counted from when it began. const SECRET_CACHE_MS = 60_000; +// How long a read still in flight is shared, counted from when it began. The +// build monitor's Candidate tab stops waiting for the collector's run list +// after 5 seconds, which can leave this read in flight with nothing keeping it +// running after the response, so Vercel can suspend the instance with it in +// flight. A later caller that shared it would receive its overdue timeout +// error, so a later caller makes its own read. Keep this no longer than that +// wait. +const SECRET_IN_FLIGHT_SHARE_MS = 5_000; const SECRET_VERSION_PATTERN = /^projects\/[a-zA-Z0-9-]+\/secrets\/[a-zA-Z0-9_-]+\/versions\/(?:[1-9][0-9]*|latest)$/; const PROVIDER_PATTERN = @@ -57,15 +66,25 @@ export function createSecretReader( access: AccessVersion, now: () => number = Date.now, ) { - const cache = new Map }>(); + const cache = new Map< + string, + { requestedAt: number; answered: boolean; value: Promise } + >(); return { async readVersion(name: string): Promise { if (!validSecretVersion(name)) { throw new Error("Invalid Google Secret Manager version resource."); } + const requestedAt = now(); const cached = cache.get(name); - if (cached && cached.expires > now()) return cached.value; + if ( + cached && + requestedAt - cached.requestedAt < + (cached.answered ? SECRET_CACHE_MS : SECRET_IN_FLIGHT_SHARE_MS) + ) { + return cached.value; + } const value = Promise.resolve() .then(() => access(name)) @@ -83,7 +102,16 @@ export function createSecretReader( "Unable to load the telemetry credential from Secret Manager.", ); }); - cache.set(name, { expires: now() + SECRET_CACHE_MS, value }); + const entry = { requestedAt, answered: false, value }; + value.then( + () => { + entry.answered = true; + }, + // The caller receives the error; this handler only keeps the + // rejection from being reported as unhandled. + () => undefined, + ); + cache.set(name, entry); return value; }, };