From 17629518ce73f7677173db93a6408b164702a238 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Thu, 15 Oct 2026 23:35:22 +0400 Subject: [PATCH] Fetch only the collector runs the build monitor shows The build monitor shows a country's 60 most recent hosted runs (STAGING_RUN_LIMIT), but loadCollectorRuns asked the telemetry collector for 200-run pages and followed every cursor, so each list request read the country's whole run history and then discarded all but 60. The cost grew with every build the collector recorded. loadCollectorRuns now takes the number of runs its caller uses and asks each page only for the runs still needed, capped at the collector's 200-run page maximum. It stops once it holds that many runs or the history ends, so the build monitor makes one limit=60 request. The collector lists runs newest first with keyset cursors, so the first 60 runs are the same ones the full read used to keep. Pagination, page validation and error handling are unchanged. The tests use an in-memory collector that follows the real listing rules: newest-first order, limits of 1 to 200, opaque cursors, and a cursor on every full page. Co-Authored-By: Claude Opus 5.5 --- frontend/lib/microcosm/build-runs.test.ts | 55 ++++- frontend/lib/microcosm/build-runs.ts | 4 +- .../test-support/fake-telemetry-collector.ts | 223 ++++++++++++++++++ .../telemetry-collector-client.test.ts | 212 ++++++++++++++++- .../microcosm/telemetry-collector-client.ts | 12 +- 5 files changed, 495 insertions(+), 11 deletions(-) create mode 100644 frontend/lib/microcosm/test-support/fake-telemetry-collector.ts diff --git a/frontend/lib/microcosm/build-runs.test.ts b/frontend/lib/microcosm/build-runs.test.ts index 2a69e669..ed575fa3 100644 --- a/frontend/lib/microcosm/build-runs.test.ts +++ b/frontend/lib/microcosm/build-runs.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { type BuildRunDocuments, @@ -6,8 +6,14 @@ import { } from "@/lib/microcosm/build-monitor"; import { type HostedRunLoaders, + loadBuildRuns, refreshHostedRun, } from "@/lib/microcosm/build-runs"; +import { + FAKE_COLLECTOR_URL, + fakeTelemetryCollector, +} from "@/lib/microcosm/test-support/fake-telemetry-collector"; +import { secretManager } from "@/lib/server/secret-manager"; const NOW = Date.parse("2026-10-05T12:00:00Z"); @@ -174,3 +180,50 @@ describe("refreshHostedRun", () => { ).rejects.toThrow("collector unavailable"); }); }); + +describe("loadBuildRuns from the collector", () => { + test("the build monitor reads only its 60 most recent collector runs", async () => { + // The zz fixture country has no staging repository, so the collector is + // the only hosted source and no Hugging Face request is made. + const collector = fakeTelemetryCollector({ + countryCode: "ZZ", + runCount: 250, + }); + const originalFetch = globalThis.fetch; + const originalEnvironment = { + MICROCOSM_TELEMETRY_COLLECTOR_URL: + process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL, + MICROCOSM_TELEMETRY_READ_SECRET_VERSION: + process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION, + }; + const readSecret = spyOn(secretManager, "readVersion").mockResolvedValue( + "secret-read-token", + ); + process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL = FAKE_COLLECTOR_URL; + process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION = + "projects/test-project/secrets/collector-read/versions/latest"; + globalThis.fetch = collector.fetch; + try { + const response = await loadBuildRuns("staging", "zz"); + + const newestRunIds = collector.runs + .slice(0, 60) + .map((run) => run.run_id); + expect(response.available).toBe(true); + expect(response.problems).toEqual([]); + expect(collector.listRequests).toHaveLength(1); + expect(collector.listRequests[0].searchParams.get("limit")).toBe("60"); + expect(collector.documentRequests.toSorted()).toEqual( + newestRunIds.toSorted(), + ); + expect(response.runs.map((run) => run.run_id)).toEqual(newestRunIds); + } finally { + globalThis.fetch = originalFetch; + readSecret.mockRestore(); + for (const [name, value] of Object.entries(originalEnvironment)) { + if (value == null) delete process.env[name]; + else process.env[name] = value; + } + } + }); +}); diff --git a/frontend/lib/microcosm/build-runs.ts b/frontend/lib/microcosm/build-runs.ts index 79d2fd1c..47936092 100644 --- a/frontend/lib/microcosm/build-runs.ts +++ b/frontend/lib/microcosm/build-runs.ts @@ -227,11 +227,11 @@ async function loadHuggingFaceStaging( } async function loadCollector(country: MicrocosmCountry): Promise { - const summaries = await loadCollectorRuns(country); + const summaries = await loadCollectorRuns(country, STAGING_RUN_LIMIT); const documents: BuildRunDocuments[] = []; const problems: BuildRunProblem[] = []; await Promise.all( - summaries.slice(0, STAGING_RUN_LIMIT).map(async (summary) => { + summaries.map(async (summary) => { const key = `collector:${country}:${summary.run_id}`; const cached = stagingRunCache.get(key); const summaryFinal = FINAL_STATUSES.has(summary.status); diff --git a/frontend/lib/microcosm/test-support/fake-telemetry-collector.ts b/frontend/lib/microcosm/test-support/fake-telemetry-collector.ts new file mode 100644 index 00000000..7528e760 --- /dev/null +++ b/frontend/lib/microcosm/test-support/fake-telemetry-collector.ts @@ -0,0 +1,223 @@ +// Test support: an in-memory telemetry collector behind a fetch stub. It +// serves `GET /v1/runs` the way the collector does (newest first, a `limit` +// between 1 and 200 that defaults to 60, an opaque `before` cursor, and a +// cursor whenever a page is full) and `GET /v1/runs/{run_id}`, and records +// every request so tests can bound what the dashboard asks for. + +export const FAKE_COLLECTOR_URL = "https://telemetry.example"; + +const COLLECTOR_DEFAULT_LIMIT = 60; +const COLLECTOR_MAX_LIMIT = 200; +const NEWEST_REGISTRATION_MS = Date.parse("2026-10-05T10:00:00Z"); + +// A type alias, not an interface, so it satisfies the client's JSON-object +// summary type. +export type FakeCollectorRunSummary = { + run_id: string; + country_code: string; + pipeline: string; + candidate_id: string | null; + release_id: string | null; + run_kind: string; + status: string; + current_stage: string; + registered_at: string; + started_at: string; + updated_at: string; + ended_at: string | null; + heartbeat_at: string | null; +}; + +export interface FakeTelemetryCollectorOptions { + countryCode: string; + runCount: number; + // Serve at most this many runs per page, with a cursor, so tests can cross + // page boundaries below the requested limit. + largestPage?: number; +} + +export interface FakeTelemetryCollector { + // Every stored run, newest first: the collector's listing order. + runs: FakeCollectorRunSummary[]; + listRequests: URL[]; + // The `next_before` cursor served for each list request, in order. + servedCursors: (string | null)[]; + documentRequests: string[]; + fetch: typeof fetch; +} + +function summaryForRun( + countryCode: string, + ageIndex: number, + runCount: number, +): FakeCollectorRunSummary { + const registeredAt = new Date( + NEWEST_REGISTRATION_MS - ageIndex * 60_000, + ).toISOString(); + const finished = ageIndex > 0; + return { + run_id: `run-${String(runCount - ageIndex).padStart(4, "0")}`, + country_code: countryCode, + pipeline: "fiscal-refresh", + candidate_id: null, + release_id: null, + run_kind: "calibration", + status: finished ? "completed" : "running", + current_stage: finished ? "complete" : "target_compilation", + registered_at: registeredAt, + started_at: registeredAt, + updated_at: registeredAt, + ended_at: finished ? registeredAt : null, + heartbeat_at: null, + }; +} + +function pageCursor(run: FakeCollectorRunSummary): string { + return Buffer.from(JSON.stringify([run.registered_at, run.run_id])).toString( + "base64url", + ); +} + +function runIdFromCursor(cursor: string): string | null { + try { + const decoded = JSON.parse(Buffer.from(cursor, "base64url").toString()); + return Array.isArray(decoded) && typeof decoded[1] === "string" + ? decoded[1] + : null; + } catch { + return null; + } +} + +function runDocuments(run: FakeCollectorRunSummary) { + return { + run_id: run.run_id, + country_code: run.country_code, + progress: { + schema_version: 2, + run_id: run.run_id, + country_code: run.country_code, + pipeline: { id: run.pipeline, version: "collector-v1" }, + started_at: run.started_at, + updated_at: run.updated_at, + status: run.status, + current_stage: run.current_stage, + }, + run_manifest: null, + calibration_progress: null, + events: [ + { + schema_version: 2, + sequence: 1, + run_id: run.run_id, + event_type: "stage", + stage_id: run.current_stage, + stage: run.current_stage, + status: run.status === "running" ? "started" : "completed", + timestamp: run.updated_at, + time: run.updated_at, + details: {}, + }, + ], + }; +} + +export function fakeTelemetryCollector({ + countryCode, + runCount, + largestPage, +}: FakeTelemetryCollectorOptions): FakeTelemetryCollector { + const runs = Array.from({ length: runCount }, (_, ageIndex) => + summaryForRun(countryCode, ageIndex, runCount), + ); + const listRequests: URL[] = []; + const servedCursors: (string | null)[] = []; + const documentRequests: string[] = []; + + function listRuns(url: URL): Response { + listRequests.push(url); + if (url.searchParams.get("country") !== countryCode) { + return Response.json({ runs: [], next_before: null }); + } + const limit = Number( + url.searchParams.get("limit") ?? COLLECTOR_DEFAULT_LIMIT, + ); + if (!Number.isInteger(limit) || limit < 1 || limit > COLLECTOR_MAX_LIMIT) { + return Response.json({ detail: "Invalid limit." }, { status: 422 }); + } + let start = 0; + const cursor = url.searchParams.get("before"); + if (cursor != null) { + const cursorRunId = runIdFromCursor(cursor); + const cursorIndex = runs.findIndex((run) => run.run_id === cursorRunId); + if (cursorIndex < 0) { + return Response.json( + { detail: "Pagination cursor is invalid." }, + { status: 422 }, + ); + } + start = cursorIndex + 1; + } + const servedSize = Math.min(limit, largestPage ?? limit); + const page = runs.slice(start, start + servedSize); + const nextBefore = + page.length === servedSize ? pageCursor(page[page.length - 1]) : null; + servedCursors.push(nextBefore); + return Response.json({ runs: page, next_before: nextBefore }); + } + + function loadRun(runId: string): Response { + documentRequests.push(runId); + const run = runs.find((candidate) => candidate.run_id === runId); + if (!run) { + return Response.json({ detail: "Run was not found." }, { status: 404 }); + } + return Response.json(runDocuments(run)); + } + + const collectorFetch = (async (input: RequestInfo | URL) => { + const url = new URL(String(input)); + if (url.origin !== FAKE_COLLECTOR_URL) { + throw new Error(`Unexpected request outside the collector: ${url.origin}`); + } + if (url.pathname === "/v1/runs") return listRuns(url); + const documentPath = /^\/v1\/runs\/([^/]+)$/.exec(url.pathname); + if (documentPath) return loadRun(decodeURIComponent(documentPath[1])); + throw new Error(`Unexpected collector path: ${url.pathname}`); + }) as typeof fetch; + + return { + runs, + listRequests, + servedCursors, + documentRequests, + fetch: collectorFetch, + }; +} + +// Every run the collector lists for a country, read the way the dashboard did +// before the listing was bounded: 200-run pages, following every cursor. +export async function entireCollectorHistory( + collector: FakeTelemetryCollector, + countryCode: string, +): Promise { + const history: FakeCollectorRunSummary[] = []; + let cursor: string | null = null; + do { + const params = new URLSearchParams({ + country: countryCode, + limit: String(COLLECTOR_MAX_LIMIT), + }); + if (cursor) params.set("before", cursor); + const response = await collector.fetch( + `${FAKE_COLLECTOR_URL}/v1/runs?${params}`, + ); + const page = (await response.json()) as { + runs: FakeCollectorRunSummary[]; + next_before: string | null; + }; + history.push(...page.runs); + cursor = page.next_before; + } while (cursor); + return history; +} diff --git a/frontend/lib/server/microcosm/telemetry-collector-client.test.ts b/frontend/lib/server/microcosm/telemetry-collector-client.test.ts index da5f6cc1..542f406c 100644 --- a/frontend/lib/server/microcosm/telemetry-collector-client.test.ts +++ b/frontend/lib/server/microcosm/telemetry-collector-client.test.ts @@ -1,4 +1,4 @@ -import { afterEach, expect, spyOn, test } from "bun:test"; +import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { secretManager } from "@/lib/server/secret-manager"; import { @@ -6,6 +6,12 @@ import { loadCollectorRun, loadCollectorRuns, } from "@/lib/server/microcosm/telemetry-collector-client"; +import { + entireCollectorHistory, + FAKE_COLLECTOR_URL, + fakeTelemetryCollector, + type FakeTelemetryCollector, +} from "@/lib/microcosm/test-support/fake-telemetry-collector"; const originalFetch = globalThis.fetch; const originalUrl = process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL; @@ -27,12 +33,59 @@ afterEach(() => { } }); +const DASHBOARD_RUN_LIMIT = 60; + function configure() { - process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL = "https://telemetry.example"; + process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL = FAKE_COLLECTOR_URL; process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION = secretVersion; readSecret.mockResolvedValue("secret-read-token"); } +function serveUsCollector( + runCount: number, + largestPage?: number, +): FakeTelemetryCollector { + configure(); + const collector = fakeTelemetryCollector({ + countryCode: "US", + runCount, + largestPage, + }); + globalThis.fetch = collector.fetch; + return collector; +} + +function requestedLimits(collector: FakeTelemetryCollector): string[] { + return collector.listRequests.map( + (request) => request.searchParams.get("limit") ?? "", + ); +} + +function runIds(runs: { run_id: string }[]): string[] { + return runs.map((run) => run.run_id); +} + +function serveListPages(pages: Response[]): URL[] { + configure(); + const requests: URL[] = []; + globalThis.fetch = (async (input) => { + requests.push(new URL(String(input))); + const page = pages.shift(); + if (!page) throw new Error("The collector was asked for an extra page."); + return page; + }) as typeof fetch; + return requests; +} + +function runSummary(runId: string) { + return { + run_id: runId, + country_code: "US", + status: "completed", + updated_at: "2026-10-01T00:00:00Z", + }; +} + test("collector configuration requires an address and secret resource", () => { delete process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL; delete process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION; @@ -44,7 +97,7 @@ test("collector configuration requires an address and secret resource", () => { expect(readSecret).not.toHaveBeenCalled(); }); -test("loads every collector list page and authenticates server-side", async () => { +test("authenticates every collector list request server-side", async () => { configure(); const calls: URL[] = []; globalThis.fetch = (async (input, init) => { @@ -79,7 +132,7 @@ test("loads every collector list page and authenticates server-side", async () = }); }) as typeof fetch; - const runs = await loadCollectorRuns("us"); + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); expect(runs.map((run) => run.run_id)).toEqual(["new", "old"]); expect(calls).toHaveLength(2); @@ -120,9 +173,158 @@ test("does not call the collector when Secret Manager access fails", async () => configure(); readSecret.mockRejectedValue(new Error("Secret Manager unavailable")); const fetchSpy = spyOn(globalThis, "fetch"); - await expect(loadCollectorRuns("us")).rejects.toThrow( + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( "Secret Manager unavailable", ); expect(fetchSpy).not.toHaveBeenCalled(); fetchSpy.mockRestore(); }); + +describe("bounded collector run listing", () => { + test("requests only the most recent runs in one bounded collector request", async () => { + const collector = serveUsCollector(500); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(runs).toHaveLength(DASHBOARD_RUN_LIMIT); + expect(collector.listRequests).toHaveLength(1); + expect(requestedLimits(collector)).toEqual(["60"]); + expect(collector.listRequests[0].searchParams.has("before")).toBe(false); + expect(collector.listRequests[0].searchParams.get("country")).toBe("US"); + }); + + test("returns the same most recent runs as reading the entire history", async () => { + const collector = serveUsCollector(250); + const history = await entireCollectorHistory(collector, "US"); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(history).toHaveLength(250); + expect(runs).toEqual(history.slice(0, DASHBOARD_RUN_LIMIT)); + }); + + test("returns every run when the collector holds fewer runs than the limit", async () => { + const collector = serveUsCollector(12); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(runIds(runs)).toEqual(runIds(collector.runs)); + expect(requestedLimits(collector)).toEqual(["60"]); + }); + + test("does not follow the cursor when exactly the limit is returned", async () => { + const collector = serveUsCollector(60); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(runIds(runs)).toEqual(runIds(collector.runs)); + expect(requestedLimits(collector)).toEqual(["60"]); + // A full page always carries a cursor, even when no older run exists. + expect(collector.servedCursors[0]).not.toBeNull(); + expect(collector.listRequests).toHaveLength(1); + }); + + test("leaves out runs older than the limit", async () => { + const collector = serveUsCollector(61); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(runIds(runs)).toEqual(runIds(collector.runs.slice(0, 60))); + expect(runIds(runs)).not.toContain(collector.runs[60].run_id); + expect(collector.listRequests).toHaveLength(1); + }); + + test("follows cursors across short pages for only the remaining runs", async () => { + const collector = serveUsCollector(500, 25); + + const runs = await loadCollectorRuns("us", DASHBOARD_RUN_LIMIT); + + expect(runIds(runs)).toEqual(runIds(collector.runs.slice(0, 60))); + expect(requestedLimits(collector)).toEqual(["60", "35", "10"]); + expect(collector.listRequests[0].searchParams.has("before")).toBe(false); + expect(collector.listRequests[1].searchParams.get("before")).toBeTruthy(); + expect(collector.listRequests[2].searchParams.get("before")).not.toBe( + collector.listRequests[1].searchParams.get("before"), + ); + }); + + test("pages past the collector's 200-run maximum for larger limits", async () => { + const collector = serveUsCollector(1000); + + const runs = await loadCollectorRuns("us", 450); + + expect(runIds(runs)).toEqual(runIds(collector.runs.slice(0, 450))); + expect(requestedLimits(collector)).toEqual(["200", "200", "50"]); + }); + + test("stops at the end of history inside a page boundary", async () => { + const collector = serveUsCollector(230); + + const runs = await loadCollectorRuns("us", 450); + + expect(runIds(runs)).toEqual(runIds(collector.runs)); + expect(requestedLimits(collector)).toEqual(["200", "200"]); + }); +}); + +describe("collector run listing errors", () => { + test("rejects a failed collector list response with its status", async () => { + serveListPages([new Response("unavailable", { status: 503 })]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector request failed (503).", + ); + }); + + test("rejects a run page without a runs array", async () => { + serveListPages([Response.json({ next_before: null })]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector run page must contain a runs array.", + ); + }); + + test("rejects a run summary without a run id", async () => { + serveListPages([ + Response.json({ + runs: [{ ...runSummary("unused"), run_id: "" }], + next_before: null, + }), + ]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector run_id must be a non-empty string.", + ); + }); + + test("rejects an invalid pagination cursor", async () => { + serveListPages([Response.json({ runs: [], next_before: 7 })]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector pagination cursor is invalid.", + ); + }); + + test("rejects a repeated pagination cursor", async () => { + serveListPages([ + Response.json({ runs: [runSummary("first")], next_before: "same" }), + Response.json({ runs: [runSummary("second")], next_before: "same" }), + ]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector pagination repeated a cursor.", + ); + }); + + test("rejects instead of returning a partial list when a later page fails", async () => { + const requests = serveListPages([ + Response.json({ runs: [runSummary("first")], next_before: "next" }), + new Response("error", { status: 500 }), + ]); + + await expect(loadCollectorRuns("us", DASHBOARD_RUN_LIMIT)).rejects.toThrow( + "Telemetry collector request failed (500).", + ); + expect(requests).toHaveLength(2); + }); +}); diff --git a/frontend/lib/server/microcosm/telemetry-collector-client.ts b/frontend/lib/server/microcosm/telemetry-collector-client.ts index 251c70fd..99583c80 100644 --- a/frontend/lib/server/microcosm/telemetry-collector-client.ts +++ b/frontend/lib/server/microcosm/telemetry-collector-client.ts @@ -119,8 +119,14 @@ function parseRunPage(value: JsonObject): CollectorRunPage { }; } +// The collector rejects a `GET /v1/runs` page larger than this. +const COLLECTOR_MAX_PAGE_SIZE = 200; + +// The country's `limit` most recent runs, newest first. Each page asks only for +// the runs still needed, so the dashboard never reads the full history. export async function loadCollectorRuns( country: MicrocosmCountry, + limit: number, ): Promise { const runs: CollectorRunSummary[] = []; const visited = new Set(); @@ -128,7 +134,7 @@ export async function loadCollectorRuns( do { const params = new URLSearchParams({ country: collectorCountry(country), - limit: "200", + limit: String(Math.min(limit - runs.length, COLLECTOR_MAX_PAGE_SIZE)), }); if (before) params.set("before", before); const page = parseRunPage(await collectorJson(`/v1/runs?${params}`)); @@ -140,8 +146,8 @@ export async function loadCollectorRuns( } visited.add(before); } - } while (before); - return runs; + } while (before && runs.length < limit); + return runs.slice(0, limit); } export async function loadCollectorRun(