From b5d8b4f9c93414642f08ee25db30151a57df6aa9 Mon Sep 17 00:00:00 2001 From: debuggingfuture Date: Fri, 23 Oct 2026 14:14:31 +0800 Subject: [PATCH] fix(cache): lane compile-cache writes per pull request and bound them at admission A compile-cache token now carries a lane: base (default-branch executions), pr- (one pull request) or read-only. A pull request writes only its own lane and reads it before base, so no pull request can plant an entry another pull request or a default-branch gate links. contextful-gate names base only from a default-branch push. Every PUT reserves its length in a per-repository CompileCacheLedger Durable Object against the repository bound and a per-execution budget; the workflow ends the execution's record when its run finishes, which refuses that token's further writes. A deploy without the binding serves reads and refuses writes. The overlay merge carries the template's Durable Object bindings and migrations into an overlay whose migrations lead the template's. The sweep resumes from a stored cursor, holds one repository's listing at a time, measures repositories it lists whole into the ledger and releases what it deletes from the rest. A GET of an entry past half its TTL rewrites it, so eviction and expiry count from last use. Only an agent run that armed the AgentBudget kills it at run end, and the command wrapper puts its closing parenthesis on its own line. Spec: specs/adr/0007-cost-efficient-execution.md --- .../dispatcher/src/compile-cache-ledger-do.ts | 31 ++ .../src/compile-cache-ledger.test.ts | 66 ++++ apps/dispatcher/src/compile-cache-ledger.ts | 103 ++++++ apps/dispatcher/src/compile-cache-token.ts | 75 ++-- apps/dispatcher/src/env.ts | 14 +- apps/dispatcher/src/index.ts | 5 +- apps/dispatcher/src/overlay-limits.test.ts | 60 ++++ apps/dispatcher/src/router.ts | 14 +- .../src/routes/compile-cache.test.ts | 297 +++++++++++++--- apps/dispatcher/src/routes/compile-cache.ts | 325 +++++++++++++----- apps/dispatcher/src/routes/scheduled.ts | 9 +- apps/dispatcher/src/run-end.test.ts | 70 ++++ apps/dispatcher/src/run-end.ts | 44 +++ apps/dispatcher/src/workflow.ts | 45 ++- infra/overlay-limits.mjs | 24 +- packages/core/src/define-run.ts | 21 +- .../core/src/primitives/compile-cache.test.ts | 21 +- packages/core/src/primitives/compile-cache.ts | 25 +- packages/core/src/primitives/index.ts | 2 + runs/check.test.ts | 16 +- runs/check.ts | 18 +- runs/contextful-gate.test.ts | 10 + runs/contextful-gate.ts | 13 +- specs/adr/0007-cost-efficient-execution.md | 47 ++- specs/adr/README.md | 2 +- wrangler.jsonc | 12 +- 26 files changed, 1137 insertions(+), 232 deletions(-) create mode 100644 apps/dispatcher/src/compile-cache-ledger-do.ts create mode 100644 apps/dispatcher/src/compile-cache-ledger.test.ts create mode 100644 apps/dispatcher/src/compile-cache-ledger.ts create mode 100644 apps/dispatcher/src/run-end.test.ts create mode 100644 apps/dispatcher/src/run-end.ts diff --git a/apps/dispatcher/src/compile-cache-ledger-do.ts b/apps/dispatcher/src/compile-cache-ledger-do.ts new file mode 100644 index 0000000..2c0670f --- /dev/null +++ b/apps/dispatcher/src/compile-cache-ledger-do.ts @@ -0,0 +1,31 @@ +// FlareDispatch Dispatcher — the CompileCacheLedger Durable Object. +// +// One instance per repository (`idFromName()`, lowercase). The +// ledger logic is compile-cache-ledger.ts; this class runs it over the +// object's storage, one call at a time. +// +// Spec: specs/adr/0007-cost-efficient-execution.md § Shared compile cache. + +import { DurableObject } from "cloudflare:workers"; +import { type CompileCacheLedgerApi, makeCompileCacheLedger, type ReserveRequest, type ReserveResult } from "./compile-cache-ledger"; +import type { Env } from "./env"; + +export class CompileCacheLedger extends DurableObject implements CompileCacheLedgerApi { + private readonly ledger = makeCompileCacheLedger(this.ctx.storage); + + async reserve(request: ReserveRequest): Promise { + return this.ledger.reserve(request); + } + + async end(request: { readonly executionId: string; readonly expiresAt: number }): Promise { + return this.ledger.end(request); + } + + async release(request: { readonly bytes: number; readonly nowSec: number }): Promise { + return this.ledger.release(request); + } + + async measure(request: { readonly bytes: number; readonly nowSec: number }): Promise { + return this.ledger.measure(request); + } +} diff --git a/apps/dispatcher/src/compile-cache-ledger.test.ts b/apps/dispatcher/src/compile-cache-ledger.test.ts new file mode 100644 index 0000000..0c61bbd --- /dev/null +++ b/apps/dispatcher/src/compile-cache-ledger.test.ts @@ -0,0 +1,66 @@ +// The compile-cache ledger: the per-repository byte count every PUT reserves +// against, and the per-execution budget and end marker that stop a token's +// writes. + +import { describe, expect, it } from "vitest"; +import { type LedgerStorage, makeCompileCacheLedger } from "./compile-cache-ledger"; + +const memoryStorage = (): LedgerStorage & { readonly map: Map } => { + const map = new Map(); + return { + map, + get: async (key: string) => map.get(key) as T | undefined, + put: async (key: string, value: T) => { + map.set(key, value); + }, + delete: async (key: string) => map.delete(key), + list: async (opts: { prefix: string }) => + new Map([...map].filter(([key]) => key.startsWith(opts.prefix)) as [string, T][]), + }; +}; + +const NOW = 1_800_000_000; +const bounds = { maxRepoBytes: 100, maxExecutionBytes: 60 }; +const reserve = (executionId: string, bytes: number) => ({ executionId, expiresAt: NOW + 3600, bytes, ...bounds }); + +describe("compile-cache ledger", () => { + it("admits writes until the repository bound and counts every admitted byte", async () => { + const ledger = makeCompileCacheLedger(memoryStorage()); + expect(await ledger.reserve(reserve("a", 50))).toEqual({ ok: true, repoBytes: 50 }); + expect(await ledger.reserve(reserve("b", 50))).toEqual({ ok: true, repoBytes: 100 }); + expect(await ledger.reserve(reserve("c", 1))).toEqual({ ok: false, reason: "repository", repoBytes: 100 }); + }); + + it("refuses an execution's write past its own budget while other executions still write", async () => { + const ledger = makeCompileCacheLedger(memoryStorage()); + expect(await ledger.reserve(reserve("a", 60))).toEqual({ ok: true, repoBytes: 60 }); + expect(await ledger.reserve(reserve("a", 1))).toEqual({ ok: false, reason: "execution", repoBytes: 60 }); + expect(await ledger.reserve(reserve("b", 10))).toEqual({ ok: true, repoBytes: 70 }); + }); + + it("refuses every write of an execution that ended, before or after its first write", async () => { + const ledger = makeCompileCacheLedger(memoryStorage()); + await ledger.reserve(reserve("a", 10)); + await ledger.end({ executionId: "a", expiresAt: NOW + 3600 }); + expect(await ledger.reserve(reserve("a", 1))).toEqual({ ok: false, reason: "ended", repoBytes: 10 }); + await ledger.end({ executionId: "never-wrote", expiresAt: NOW + 3600 }); + expect(await ledger.reserve(reserve("never-wrote", 1))).toEqual({ ok: false, reason: "ended", repoBytes: 10 }); + }); + + it("subtracts released bytes, never below zero", async () => { + const ledger = makeCompileCacheLedger(memoryStorage()); + await ledger.reserve(reserve("a", 40)); + expect(await ledger.release({ bytes: 30, nowSec: NOW })).toBe(10); + expect(await ledger.release({ bytes: 30, nowSec: NOW })).toBe(0); + }); + + it("replaces the count with a measured size and forgets executions whose tokens expired", async () => { + const storage = memoryStorage(); + const ledger = makeCompileCacheLedger(storage); + await ledger.reserve({ ...reserve("old", 10), expiresAt: NOW - 1 }); + await ledger.reserve(reserve("live", 10)); + expect(await ledger.measure({ bytes: 75, nowSec: NOW })).toBe(75); + expect([...storage.map.keys()].filter((key) => key.startsWith("x:"))).toEqual(["x:live"]); + expect(await ledger.reserve(reserve("live", 25))).toEqual({ ok: true, repoBytes: 100 }); + }); +}); diff --git a/apps/dispatcher/src/compile-cache-ledger.ts b/apps/dispatcher/src/compile-cache-ledger.ts new file mode 100644 index 0000000..57a91ea --- /dev/null +++ b/apps/dispatcher/src/compile-cache-ledger.ts @@ -0,0 +1,103 @@ +// FlareDispatch Dispatcher — the compile-cache ledger. +// +// One ledger per repository (the `CompileCacheLedger` Durable Object, +// compile-cache-ledger-do.ts, addressed by `idFromName()`) holds: +// +// repo the repository's bytes: every admitted PUT adds its length, +// the sweep subtracts what it deletes and, after listing the +// whole repository, replaces the count with the measured size +// x: one execution's admitted bytes, whether it ended, and when +// its token expires +// +// A PUT reserves its length here before it writes, so the repository bound +// and the per-execution budget hold at admission rather than as of the last +// sweep. The workflow ends an execution's record when the run finishes, which +// refuses that token's further writes. The Durable Object runs one call at a +// time, so two reserves never both pass on the same headroom. +// +// An overwritten key counts twice until the next whole-repository sweep; the +// count errs high, never low. +// +// Spec: specs/adr/0007-cost-efficient-execution.md § Shared compile cache. + +/** The storage the ledger reads and writes: a Durable Object's `ctx.storage`. */ +export type LedgerStorage = { + get(key: string): Promise; + put(key: string, value: T): Promise; + delete(key: string): Promise; + list(options: { prefix: string }): Promise>; +}; + +type ExecutionRecord = { readonly bytes: number; readonly ended: boolean; readonly expiresAt: number }; + +const REPO_KEY = "repo"; +const EXECUTION_PREFIX = "x:"; + +export type ReserveRequest = { + readonly executionId: string; + /** The token's expiry, epoch seconds: the record is kept until then. */ + readonly expiresAt: number; + readonly bytes: number; + readonly maxRepoBytes: number; + readonly maxExecutionBytes: number; +}; + +/** A reserve's verdict, with the repository's count after it. */ +export type ReserveResult = + | { readonly ok: true; readonly repoBytes: number } + | { readonly ok: false; readonly reason: "ended" | "execution" | "repository"; readonly repoBytes: number }; + +/** The ledger's calls, as the route and the sweep make them. */ +export type CompileCacheLedgerApi = { + reserve(request: ReserveRequest): Promise; + end(request: { readonly executionId: string; readonly expiresAt: number }): Promise; + release(request: { readonly bytes: number; readonly nowSec: number }): Promise; + measure(request: { readonly bytes: number; readonly nowSec: number }): Promise; +}; + +/** The ledger over `storage`. */ +export const makeCompileCacheLedger = (storage: LedgerStorage): CompileCacheLedgerApi => { + const repoBytes = async (): Promise => (await storage.get(REPO_KEY)) ?? 0; + const forgetExpired = async (nowSec: number): Promise => { + const records = await storage.list({ prefix: EXECUTION_PREFIX }); + for (const [key, record] of records) if (record.expiresAt <= nowSec) await storage.delete(key); + }; + return { + reserve: async (request) => { + const key = `${EXECUTION_PREFIX}${request.executionId}`; + const execution = (await storage.get(key)) ?? { + bytes: 0, + ended: false, + expiresAt: request.expiresAt, + }; + const repo = await repoBytes(); + if (execution.ended) return { ok: false, reason: "ended", repoBytes: repo }; + if (execution.bytes + request.bytes > request.maxExecutionBytes) + return { ok: false, reason: "execution", repoBytes: repo }; + if (repo + request.bytes > request.maxRepoBytes) return { ok: false, reason: "repository", repoBytes: repo }; + await storage.put(key, { ...execution, bytes: execution.bytes + request.bytes }); + await storage.put(REPO_KEY, repo + request.bytes); + return { ok: true, repoBytes: repo + request.bytes }; + }, + end: async (request) => { + const key = `${EXECUTION_PREFIX}${request.executionId}`; + const execution = await storage.get(key); + await storage.put(key, { + bytes: execution?.bytes ?? 0, + ended: true, + expiresAt: Math.max(execution?.expiresAt ?? 0, request.expiresAt), + }); + }, + release: async (request) => { + const next = Math.max(0, (await repoBytes()) - request.bytes); + await storage.put(REPO_KEY, next); + await forgetExpired(request.nowSec); + return next; + }, + measure: async (request) => { + await storage.put(REPO_KEY, request.bytes); + await forgetExpired(request.nowSec); + return request.bytes; + }, + }; +}; diff --git a/apps/dispatcher/src/compile-cache-token.ts b/apps/dispatcher/src/compile-cache-token.ts index 89d9574..43fd6e0 100644 --- a/apps/dispatcher/src/compile-cache-token.ts +++ b/apps/dispatcher/src/compile-cache-token.ts @@ -1,13 +1,19 @@ // FlareDispatch Dispatcher — compile-cache tokens. // // The compile-cache route (routes/compile-cache.ts) admits a request carrying -// a token minted for one repository and one execution, valid until that -// execution's ceiling: +// a token minted for one repository, one execution and one lane, valid until +// that execution's ceiling: // -// token = "cc1." "." base64url(executionId) "." mac -// mac = base64url( HMAC-SHA256( k_cc, "cc1\n" repo "\n" executionId "\n" expiresAt ) ) +// token = "cc2." "." base64url(executionId) "." lane "." mac +// mac = base64url( HMAC-SHA256( k_cc, "cc2\n" repo "\n" executionId "\n" lane "\n" expiresAt ) ) // k_cc = HKDF-SHA256( HMAC_SECRET, salt="", info="flare-dispatch/compile-cache/v1" ) // +// `lane` (primitives/compile-cache.ts `CompileCacheLane`) decides where the +// token's writes land and which entries it reads: `base`, `pr-` or +// `read-only`. The run definition names it from the execution's trigger; a +// dispatch caller able to name `base` holds `HMAC_SECRET`, the key this MAC +// derives from, so the lane adds no authority the caller lacks. +// // `repo` is lowercase `owner/name` and is not in the token: the route takes it // from the request path, so a token presented under another repository's // prefix fails the MAC. `expiresAt` is epoch seconds. The HKDF label keeps @@ -22,12 +28,14 @@ import { COMPILE_CACHE_ENDPOINT_KEY, COMPILE_CACHE_TOKEN_KEY, + type CompileCacheLane, + isCompileCacheLane, } from "@fractalboxdev/flare-dispatch-core/primitives"; import { deriveSecret } from "./capability-token"; import type { Env } from "./env"; const HKDF_INFO = "flare-dispatch/compile-cache/v1"; -const VERSION = "cc1"; +const VERSION = "cc2"; const encoder = new TextEncoder(); @@ -51,10 +59,11 @@ const fromBase64url = (text: string): string | undefined => { /** The execution id charset (log-token.ts `EXECUTION_ID_RE`). */ const EXECUTION_ID = /^[A-Za-z0-9:_-]{3,128}$/; -/** The token's scope: one repository, one execution, until `expiresAt` (epoch seconds). */ +/** The token's scope: one repository, one execution, one lane, until `expiresAt` (epoch seconds). */ export type CompileCacheScope = { readonly repo: string; readonly executionId: string; + readonly lane: CompileCacheLane; readonly expiresAt: number; }; @@ -68,7 +77,9 @@ const macKey = async (ikm: string): Promise => ); const message = (scope: CompileCacheScope): Uint8Array => - encoder.encode(`${VERSION}\n${scope.repo.toLowerCase()}\n${scope.executionId}\n${scope.expiresAt}`); + encoder.encode( + `${VERSION}\n${scope.repo.toLowerCase()}\n${scope.executionId}\n${scope.lane}\n${scope.expiresAt}`, + ); /** The key material: the Worker's `HMAC_SECRET`; `undefined` when unset (the route then refuses). */ export const resolveCompileCacheSecret = (env: Pick): string | undefined => @@ -77,10 +88,11 @@ export const resolveCompileCacheSecret = (env: Pick): string /** Mint the token for `scope`. */ export const signCompileCacheToken = async (ikm: string, scope: CompileCacheScope): Promise => { if (!EXECUTION_ID.test(scope.executionId)) throw new Error("compile-cache token: invalid execution id"); + if (!isCompileCacheLane(scope.lane)) throw new Error("compile-cache token: invalid lane"); if (!Number.isSafeInteger(scope.expiresAt) || scope.expiresAt <= 0) throw new Error("compile-cache token: invalid expiry"); const mac = new Uint8Array(await crypto.subtle.sign("HMAC", await macKey(ikm), message(scope))); - return `${VERSION}.${scope.expiresAt}.${base64url(encoder.encode(scope.executionId))}.${base64url(mac)}`; + return `${VERSION}.${scope.expiresAt}.${base64url(encoder.encode(scope.executionId))}.${scope.lane}.${base64url(mac)}`; }; /** Why a presented token was refused. */ @@ -97,14 +109,20 @@ export const verifyCompileCacheToken = async ( nowSec: number, ): Promise<{ readonly ok: true; readonly scope: CompileCacheScope } | { readonly ok: false; readonly reason: CompileCacheTokenRefusal }> => { const parts = presented.split("."); - if (parts.length !== 4 || parts[0] !== VERSION || !/^[1-9][0-9]{0,15}$/.test(parts[1]!)) + if (parts.length !== 5 || parts[0] !== VERSION || !/^[1-9][0-9]{0,15}$/.test(parts[1]!)) return { ok: false, reason: "malformed" }; const expiresAt = Number(parts[1]); const executionId = fromBase64url(parts[2]!); - if (executionId === undefined || !EXECUTION_ID.test(executionId) || !/^[A-Za-z0-9_-]{43}$/.test(parts[3]!)) + const lane = parts[3]!; + if ( + executionId === undefined || + !EXECUTION_ID.test(executionId) || + !isCompileCacheLane(lane) || + !/^[A-Za-z0-9_-]{43}$/.test(parts[4]!) + ) return { ok: false, reason: "malformed" }; - const mac = Uint8Array.from(atob(parts[3]!.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0)); - const scope = { repo, executionId, expiresAt }; + const mac = Uint8Array.from(atob(parts[4]!.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0)); + const scope = { repo, executionId, lane, expiresAt }; const valid = await crypto.subtle.verify("HMAC", await macKey(ikm), mac, message(scope)); if (!valid) return { ok: false, reason: "invalid" }; if (nowSec >= expiresAt) return { ok: false, reason: "expired" }; @@ -113,8 +131,8 @@ export const verifyCompileCacheToken = async ( /** What the workflow knows when it decides whether to mint. */ export type CompileCacheMint = { - /** The run's own answer (`RunSpec.compileCache`). */ - readonly wanted: boolean; + /** The run's own answer (`RunSpec.compileCache`): its lane, or `undefined` for no cache. */ + readonly lane: CompileCacheLane | undefined; readonly repo: string; readonly executionId: string; /** The Workflow instance's creation time, epoch ms: stable across invocations. */ @@ -125,23 +143,34 @@ export type CompileCacheMint = { readonly secret: string | undefined; }; +/** A minted grant: the config overrides that carry it, its lane, and its token's expiry (epoch seconds). */ +export type CompileCacheGrant = { + readonly overrides: Record; + readonly lane: CompileCacheLane; + readonly expiresAt: number; +}; + /** - * The config overrides that hand a run its compile-cache endpoint and token, - * or `undefined` when the run does not want the cache or the deploy has no + * The grant that hands a run its compile-cache endpoint and token, or + * `undefined` when the run does not want the cache or the deploy has no * public origin or signing secret. */ -export const compileCacheOverrides = async ( - mint: CompileCacheMint, -): Promise | undefined> => { - if (!mint.wanted || mint.publicOrigin === undefined || mint.publicOrigin === "" || mint.secret === undefined) +export const compileCacheOverrides = async (mint: CompileCacheMint): Promise => { + if (mint.lane === undefined || mint.publicOrigin === undefined || mint.publicOrigin === "" || mint.secret === undefined) return undefined; + const expiresAt = Math.floor(mint.createdAtMs / 1000) + Math.ceil(mint.ceilingSec); const token = await signCompileCacheToken(mint.secret, { repo: mint.repo, executionId: mint.executionId, - expiresAt: Math.floor(mint.createdAtMs / 1000) + Math.ceil(mint.ceilingSec), + lane: mint.lane, + expiresAt, }); return { - [COMPILE_CACHE_ENDPOINT_KEY]: `${mint.publicOrigin.replace(/\/$/, "")}/v1/compile-cache`, - [COMPILE_CACHE_TOKEN_KEY]: token, + overrides: { + [COMPILE_CACHE_ENDPOINT_KEY]: `${mint.publicOrigin.replace(/\/$/, "")}/v1/compile-cache`, + [COMPILE_CACHE_TOKEN_KEY]: token, + }, + lane: mint.lane, + expiresAt, }; }; diff --git a/apps/dispatcher/src/env.ts b/apps/dispatcher/src/env.ts index 334dc18..e417ee5 100644 --- a/apps/dispatcher/src/env.ts +++ b/apps/dispatcher/src/env.ts @@ -335,11 +335,13 @@ export interface Env { /** * Compile-cache bounds (routes/compile-cache.ts): one entry's bytes, one - * repository prefix's bytes, and the entry age in days the sweep deletes. - * Unset or malformed → 100 MiB, 20 GiB, 14 days. + * repository's bytes, one execution's written bytes, and the entry age in + * days the sweep deletes. Unset or malformed → 100 MiB, 20 GiB, 10 GiB, + * 14 days. */ readonly COMPILE_CACHE_MAX_OBJECT_BYTES?: string; readonly COMPILE_CACHE_MAX_REPO_BYTES?: string; + readonly COMPILE_CACHE_MAX_EXECUTION_BYTES?: string; readonly COMPILE_CACHE_TTL_DAYS?: string; /** @@ -431,6 +433,14 @@ export interface Env { */ readonly AGENT_BUDGET?: DurableObjectNamespace; + /** + * One compile-cache ledger per repository (compile-cache-ledger.ts): the + * byte count every compile-cache PUT reserves against, and each + * execution's write budget and end marker. Absent → the compile cache + * serves reads and refuses every write. + */ + readonly COMPILE_CACHE_LEDGER?: DurableObjectNamespace; + /** * Optional dedicated secret for the agent model-proxy capability token. Falls * back to `HMAC_SECRET` (HKDF label keeps them independent). Neither set → the diff --git a/apps/dispatcher/src/index.ts b/apps/dispatcher/src/index.ts index 52b623f..a59517e 100644 --- a/apps/dispatcher/src/index.ts +++ b/apps/dispatcher/src/index.ts @@ -28,10 +28,11 @@ import { RunSandbox, RunSandboxBrowser, RunSandboxAgent, RunSandboxRelease } fro // Re-export the binding classes so wrangler's `main` entry resolves them. export { RunWorkflow } from "./workflow"; export { AgentBudget } from "./agent-budget-do"; +export { CompileCacheLedger } from "./compile-cache-ledger-do"; export { RunSandbox, RunSandboxBrowser, RunSandboxAgent, RunSandboxRelease }; export default { - async fetch(request: Request, env: Env): Promise { + async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { // Preview-URL proxying for `exposePort` (the `cdp-acceptance` reachability // path): a request to `--.` // is forwarded to that container's port. `proxyToSandbox` returns `null` @@ -49,7 +50,7 @@ export default { Sandbox: env.RUNS_SANDBOX, }); if (proxied !== null) return proxied; - return handleRequest(request, env); + return handleRequest(request, env, ctx); }, /** * Cron Trigger entry — fires once per `triggers.crons` cadence. We hand off diff --git a/apps/dispatcher/src/overlay-limits.test.ts b/apps/dispatcher/src/overlay-limits.test.ts index e7c2cb3..a29b091 100644 --- a/apps/dispatcher/src/overlay-limits.test.ts +++ b/apps/dispatcher/src/overlay-limits.test.ts @@ -79,4 +79,64 @@ describe("mergeOverlayLimits", () => { { class_name: "RunSandboxBrowser" }, ]); }); + + describe("Durable Objects", () => { + type Config = { + durable_objects?: { bindings: { name: string; class_name: string }[] }; + migrations?: { tag: string; new_sqlite_classes?: string[] }[]; + }; + const tpl = `{ + "durable_objects": { "bindings": [ + { "name": "AGENT_BUDGET", "class_name": "AgentBudget" }, + { "name": "COMPILE_CACHE_LEDGER", "class_name": "CompileCacheLedger" }, + ] }, + "migrations": [ + { "tag": "v0", "new_sqlite_classes": ["AgentBudget"] }, + { "tag": "v1", "new_sqlite_classes": ["CompileCacheLedger"] }, + ], + }`; + + it("adds a template binding the overlay lacks, with its migration, when the overlay's migrations lead the template's", () => { + const own = `{ "name": "x", + "durable_objects": { "bindings": [{ "name": "AGENT_BUDGET", "class_name": "AgentBudget" }] }, + "migrations": [{ "tag": "v0", "new_sqlite_classes": ["AgentBudget"] }] }`; + const merged = parseJsonc(mergeOverlayLimits(tpl, own)) as Config; + expect(merged.durable_objects?.bindings.map((binding) => binding.name)).toEqual(["AGENT_BUDGET", "COMPILE_CACHE_LEDGER"]); + expect(merged.migrations?.map((migration) => migration.tag)).toEqual(["v0", "v1"]); + }); + + it("leaves an overlay whose migrations diverge from the template's unchanged", () => { + const own = `{ "name": "x", + "durable_objects": { "bindings": [{ "name": "AGENT_BUDGET", "class_name": "AgentBudget" }] }, + "migrations": [{ "tag": "own-v0", "new_sqlite_classes": ["AgentBudget"] }] }`; + expect(parseJsonc(mergeOverlayLimits(tpl, own))).toEqual(parseJsonc(own)); + }); + + it("keeps a binding the overlay already declares", () => { + const own = `{ "name": "x", + "durable_objects": { "bindings": [ + { "name": "AGENT_BUDGET", "class_name": "AgentBudget" }, + { "name": "COMPILE_CACHE_LEDGER", "class_name": "CompileCacheLedger" }, + ] }, + "migrations": [ + { "tag": "v0", "new_sqlite_classes": ["AgentBudget"] }, + { "tag": "v1", "new_sqlite_classes": ["CompileCacheLedger"] }, + ] }`; + expect(parseJsonc(mergeOverlayLimits(tpl, own))).toEqual(parseJsonc(own)); + }); + + it("carries the committed wrangler.jsonc's ledger into an overlay that predates it", () => { + const committed = parseJsonc(template) as Config; + const own = JSON.stringify({ + name: "x", + durable_objects: { + bindings: committed.durable_objects!.bindings.filter((binding) => binding.name !== "COMPILE_CACHE_LEDGER"), + }, + migrations: committed.migrations!.filter((migration) => !migration.new_sqlite_classes?.includes("CompileCacheLedger")), + }); + const merged = parseJsonc(mergeOverlayLimits(template, own)) as Config; + expect(merged.durable_objects?.bindings).toContainEqual({ name: "COMPILE_CACHE_LEDGER", class_name: "CompileCacheLedger" }); + expect(merged.migrations).toEqual(committed.migrations); + }); + }); }); diff --git a/apps/dispatcher/src/router.ts b/apps/dispatcher/src/router.ts index 76bbb47..85dc169 100644 --- a/apps/dispatcher/src/router.ts +++ b/apps/dispatcher/src/router.ts @@ -28,7 +28,11 @@ const json = (body: unknown, status: number): Response => }); /** Route an inbound request to its handler. */ -export const handleRequest = async (request: Request, env: Env): Promise => { +export const handleRequest = async ( + request: Request, + env: Env, + ctx?: Pick, +): Promise => { const url = new URL(request.url); const segments = url.pathname.split("/").filter((s) => s.length > 0); @@ -51,9 +55,13 @@ export const handleRequest = async (request: Request, env: Env): Promise) => ctx.waitUntil(promise) } : {}), + }); } return handleViaApp(request, env); diff --git a/apps/dispatcher/src/routes/compile-cache.test.ts b/apps/dispatcher/src/routes/compile-cache.test.ts index 26f1c34..645c72b 100644 --- a/apps/dispatcher/src/routes/compile-cache.test.ts +++ b/apps/dispatcher/src/routes/compile-cache.test.ts @@ -1,12 +1,14 @@ // `/v1/compile-cache/…` — the WebDAV subset sccache's backend issues, its // token scope, its bounds, and the sweep that holds each prefix to its ceiling. -import { describe, expect, it } from "vitest"; +import type { CompileCacheLane } from "@fractalboxdev/flare-dispatch-core/primitives"; +import { beforeEach, describe, expect, it } from "vitest"; import { compileCacheOverrides, signCompileCacheToken, verifyCompileCacheToken, } from "../compile-cache-token"; +import { type LedgerStorage, makeCompileCacheLedger } from "../compile-cache-ledger"; import type { Env } from "../env"; import { handleRequest } from "../router"; import { makeFakeEnv, makeFakeR2, makeFakeWorkflow } from "../test-helpers"; @@ -15,8 +17,8 @@ import { handleCompileCache, parseCompileCachePath, resolveCompileCacheLimits, + SWEEP_STATE_KEY, sweepCompileCache, - usageKey, } from "./compile-cache"; const SECRET = "compile-cache-secret-aaaaaaaaaaaaaaaa"; @@ -56,14 +58,17 @@ const makeBucket = (pageSize = 1000) => { for (const key of Array.isArray(keys) ? keys : [keys]) store.delete(key); }, list: async (opts: { prefix?: string; cursor?: string; limit?: number }) => { - const keys = [...store.keys()].filter((key) => key.startsWith(opts.prefix ?? "")).sort(); - const start = opts.cursor === undefined ? 0 : Number(opts.cursor); - const page = keys.slice(start, start + Math.min(pageSize, opts.limit ?? 1000)); - const next = start + page.length; + // The cursor names the last key returned, so deletions between pages + // and between sweeps shift nothing, as with R2's opaque cursor. + const keys = [...store.keys()] + .filter((key) => key.startsWith(opts.prefix ?? "") && (opts.cursor === undefined || key > opts.cursor)) + .sort(); + const page = keys.slice(0, Math.min(pageSize, opts.limit ?? 1000)); + const more = page.length < keys.length; return { objects: page.map((key) => meta(key, store.get(key)!)), - truncated: next < keys.length, - ...(next < keys.length ? { cursor: String(next) } : {}), + truncated: more, + ...(more ? { cursor: page[page.length - 1] } : {}), }; }, } as unknown as CompileCacheBucket; @@ -72,16 +77,51 @@ const makeBucket = (pageSize = 1000) => { return { bucket, store, seed }; }; +const memoryStorage = (): LedgerStorage => { + const map = new Map(); + return { + get: async (key: string) => map.get(key) as T | undefined, + put: async (key: string, value: T) => { + map.set(key, value); + }, + delete: async (key: string) => map.delete(key), + list: async (opts: { prefix: string }) => + new Map([...map].filter(([key]) => key.startsWith(opts.prefix)) as [string, T][]), + }; +}; + +/** One in-memory ledger per repository, addressed the way the route addresses the Durable Object. */ +const makeLedgers = () => { + const byName = new Map>(); + const ledger = (name: string) => { + let found = byName.get(name); + if (found === undefined) { + found = makeCompileCacheLedger(memoryStorage()); + byName.set(name, found); + } + return found; + }; + const namespace = { idFromName: (name: string) => name, get: (id: string) => ledger(id) }; + return { ledger, namespace: namespace as unknown as Env["COMPILE_CACHE_LEDGER"] }; +}; + +let ledgers = makeLedgers(); +beforeEach(() => { + ledgers = makeLedgers(); +}); + const envWith = (bucket: CompileCacheBucket, over: Partial = {}) => ({ HMAC_SECRET: SECRET, RUNS_STORAGE: bucket, + COMPILE_CACHE_LEDGER: ledgers.namespace, ...over, }); -const token = (over: { repo?: string; expiresAt?: number } = {}) => +const token = (over: { repo?: string; expiresAt?: number; lane?: CompileCacheLane; executionId?: string } = {}) => signCompileCacheToken(SECRET, { repo: over.repo ?? "owner/name", - executionId: EXEC, + executionId: over.executionId ?? EXEC, + lane: over.lane ?? "base", expiresAt: over.expiresAt ?? NOW + 3600, }); @@ -89,7 +129,13 @@ const call = async ( bucket: CompileCacheBucket, method: string, url: string, - opts: { token?: string | null; body?: string; headers?: Record; env?: Partial } = {}, + opts: { + token?: string | null; + body?: string; + headers?: Record; + env?: Partial; + waitUntil?: (promise: Promise) => void; + } = {}, ) => { const bearer = opts.token === undefined ? await token() : opts.token; const headers: Record = { ...opts.headers }; @@ -99,7 +145,10 @@ const call = async ( return handleCompileCache( new Request(url, { method, headers, ...(opts.body !== undefined ? { body: opts.body } : {}) }), envWith(bucket, opts.env), - { now: () => NOW_MS }, + { + now: () => NOW_MS, + ...(opts.waitUntil !== undefined ? { waitUntil: opts.waitUntil, fixedLength: (stream: ReadableStream) => stream } : {}), + }, ); }; @@ -124,11 +173,19 @@ describe("compile-cache auth", () => { it("refuses a token whose expiry was rewritten", async () => { const minted = await token(); - const [v, , exec, mac] = minted.split("."); - const res = await call(makeBucket().bucket, "GET", `${BASE}/a/b/c/abc`, { token: `${v}.${NOW + 999_999}.${exec}.${mac}` }); + const [v, , exec, lane, mac] = minted.split("."); + const res = await call(makeBucket().bucket, "GET", `${BASE}/a/b/c/abc`, { token: `${v}.${NOW + 999_999}.${exec}.${lane}.${mac}` }); expect(res.status).toBe(403); }); + it("refuses a token whose lane was rewritten", async () => { + const minted = await token({ lane: "pr-7" }); + const [v, expiry, exec, , mac] = minted.split("."); + const res = await call(makeBucket().bucket, "PUT", `${BASE}/a/b/c/abc`, { token: `${v}.${expiry}.${exec}.base.${mac}`, body: "x" }); + expect(res.status).toBe(403); + expect(await res.text()).toBe("token invalid"); + }); + it("refuses a malformed token", async () => { const res = await call(makeBucket().bucket, "GET", `${BASE}/a/b/c/abc`, { token: "not-a-token" }); expect(res.status).toBe(403); @@ -148,7 +205,7 @@ describe("compile-cache auth", () => { it("verifies a minted token's scope", async () => { const verdict = await verifyCompileCacheToken(SECRET, "owner/name", await token(), NOW); - expect(verdict).toEqual({ ok: true, scope: { repo: "owner/name", executionId: EXEC, expiresAt: NOW + 3600 } }); + expect(verdict).toEqual({ ok: true, scope: { repo: "owner/name", executionId: EXEC, lane: "base", expiresAt: NOW + 3600 } }); }); }); @@ -157,7 +214,7 @@ describe("compile-cache WebDAV", () => { const { bucket, store } = makeBucket(); const put = await call(bucket, "PUT", `${BASE}/a/b/c/abcdef`, { body: "rlib-bytes" }); expect(put.status).toBe(201); - expect([...store.keys()]).toEqual(["compile-cache/owner/name/1.90.0/x86_64-unknown-linux-gnu/a/b/c/abcdef"]); + expect([...store.keys()]).toEqual(["compile-cache/owner/name/@base/1.90.0/x86_64-unknown-linux-gnu/a/b/c/abcdef"]); const get = await call(bucket, "GET", `${BASE}/a/b/c/abcdef`); expect(get.status).toBe(200); @@ -214,15 +271,49 @@ describe("compile-cache WebDAV", () => { expect(res.status).toBe(411); }); - it("refuses a PUT once the measured prefix plus the entry exceeds the repository bound", async () => { + it("refuses a PUT once the repository's admitted bytes plus the entry exceed its bound", async () => { const { bucket, store } = makeBucket(); - await bucket.put(usageKey("owner/name"), "", { customMetadata: { bytes: "95" } }); - const res = await call(bucket, "PUT", `${BASE}/a/b/c/x`, { - body: "0123456789", - env: { COMPILE_CACHE_MAX_REPO_BYTES: "100" }, - }); + const env = { COMPILE_CACHE_MAX_REPO_BYTES: "15" }; + expect((await call(bucket, "PUT", `${BASE}/a/b/c/x`, { body: "0123456789", env })).status).toBe(201); + const other = await token({ executionId: "check:owner_name:other" }); + const res = await call(bucket, "PUT", `${BASE}/a/b/c/y`, { token: other, body: "0123456789", env }); expect(res.status).toBe(507); - expect([...store.keys()]).toEqual([usageKey("owner/name")]); + expect(store.size).toBe(1); + }); + + it("counts every PUT at once, so writes between sweeps cannot pass the repository bound", async () => { + const { bucket, store } = makeBucket(); + const env = { COMPILE_CACHE_MAX_REPO_BYTES: "25" }; + const statuses: number[] = []; + for (let i = 0; i < 5; i++) statuses.push((await call(bucket, "PUT", `${BASE}/a/b/c/k${i}`, { body: "0123456789", env })).status); + expect(statuses).toEqual([201, 201, 507, 507, 507]); + expect(store.size).toBe(2); + }); + + it("refuses an execution's PUT past its own byte budget", async () => { + const { bucket } = makeBucket(); + const env = { COMPILE_CACHE_MAX_EXECUTION_BYTES: "15" }; + expect((await call(bucket, "PUT", `${BASE}/a/b/c/x`, { body: "0123456789", env })).status).toBe(201); + const res = await call(bucket, "PUT", `${BASE}/a/b/c/y`, { body: "0123456789", env }); + expect(res.status).toBe(507); + expect(await res.text()).toContain("execution"); + }); + + it("refuses a PUT once the execution ended, though its token has not expired", async () => { + const { bucket } = makeBucket(); + await ledgers.ledger("owner/name").end({ executionId: EXEC, expiresAt: NOW + 3600 }); + const res = await call(bucket, "PUT", `${BASE}/a/b/c/x`, { body: "0123456789" }); + expect(res.status).toBe(403); + expect(await res.text()).toBe("execution ended"); + expect((await call(bucket, "GET", `${BASE}/a/b/c/x`)).status).toBe(404); + }); + + it("refuses every PUT on a deploy without the ledger, and still serves reads", async () => { + const { bucket, seed } = makeBucket(); + seed("compile-cache/owner/name/@base/1.90.0/x86_64-unknown-linux-gnu/a/b/c/x", 3, NOW_MS); + const env = { COMPILE_CACHE_LEDGER: undefined }; + expect((await call(bucket, "PUT", `${BASE}/a/b/c/y`, { body: "0123456789", env })).status).toBe(503); + expect((await call(bucket, "GET", `${BASE}/a/b/c/x`, { env })).status).toBe(200); }); it("rejects paths that escape or skip the repository", () => { @@ -233,14 +324,15 @@ describe("compile-cache WebDAV", () => { expect(parseCompileCachePath("/v1/compile-cache/owner")).toBeUndefined(); expect(parseCompileCachePath("/v1/compile-cache/owner/name/")).toEqual({ repo: "owner/name", - key: "compile-cache/owner/name/", + key: "", collection: true, }); expect(parseCompileCachePath("/v1/compile-cache/owner/name/.sccache_check")).toEqual({ repo: "owner/name", - key: "compile-cache/owner/name/.sccache_check", + key: ".sccache_check", collection: false, }); + expect(parseCompileCachePath("/v1/compile-cache/owner/name/@base/x")).toBeUndefined(); }); it("is reachable through the dispatcher router", async () => { @@ -250,11 +342,75 @@ describe("compile-cache WebDAV", () => { }); }); +describe("compile-cache lanes", () => { + const entry = "1.90.0/x86_64-unknown-linux-gnu/a/b/c/abcdef"; + + it("lands a pull request's PUT in its own lane, which no other lane reads", async () => { + const { bucket, store } = makeBucket(); + const pr7 = await token({ lane: "pr-7" }); + expect((await call(bucket, "PUT", `${BASE}/a/b/c/abcdef`, { token: pr7, body: "planted" })).status).toBe(201); + expect([...store.keys()]).toEqual([`compile-cache/owner/name/@pr-7/${entry}`]); + expect((await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: await token() })).status).toBe(404); + expect((await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: await token({ lane: "pr-8" }) })).status).toBe(404); + expect((await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: await token({ lane: "read-only" }) })).status).toBe(404); + expect(await (await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: pr7 })).text()).toBe("planted"); + }); + + it("serves a pull request the default branch's entry, and its own entry over it", async () => { + const { bucket } = makeBucket(); + const pr7 = await token({ lane: "pr-7" }); + await call(bucket, "PUT", `${BASE}/a/b/c/abcdef`, { body: "from-base" }); + expect(await (await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: pr7 })).text()).toBe("from-base"); + expect((await call(bucket, "HEAD", `${BASE}/a/b/c/abcdef`, { token: pr7 })).status).toBe(200); + expect((await call(bucket, "PROPFIND", `${BASE}/a/b/c/abcdef`, { token: pr7, headers: { depth: "0" } })).status).toBe(207); + await call(bucket, "PUT", `${BASE}/a/b/c/abcdef`, { token: pr7, body: "from-pr" }); + expect(await (await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { token: pr7 })).text()).toBe("from-pr"); + expect(await (await call(bucket, "GET", `${BASE}/a/b/c/abcdef`)).text()).toBe("from-base"); + }); + + it("refuses a PUT from a read-only token", async () => { + const { bucket, store } = makeBucket(); + const res = await call(bucket, "PUT", `${BASE}/a/b/c/abcdef`, { token: await token({ lane: "read-only" }), body: "x" }); + expect(res.status).toBe(403); + expect(await res.text()).toBe("token grants no writes"); + expect(store.size).toBe(0); + }); + + it("rewrites an entry read past half its lifetime, so the sweep ages entries by last use", async () => { + const { bucket, store, seed } = makeBucket(); + const DAY = 24 * 3600 * 1000; + const key = `compile-cache/owner/name/@base/${entry}`; + seed(key, 4, NOW_MS - 8 * DAY); + const pending: Promise[] = []; + const res = await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { + token: await token({ lane: "pr-7" }), + waitUntil: (promise) => pending.push(promise), + }); + expect(res.status).toBe(200); + expect((await res.arrayBuffer()).byteLength).toBe(4); + await Promise.all(pending); + expect(store.get(key)?.uploaded.getTime()).toBe(NOW_MS); + expect([...store.keys()]).toEqual([key]); + }); + + it("leaves an entry read within half its lifetime as written", async () => { + const { bucket, store, seed } = makeBucket(); + const DAY = 24 * 3600 * 1000; + const key = `compile-cache/owner/name/@base/${entry}`; + seed(key, 4, NOW_MS - 6 * DAY); + const pending: Promise[] = []; + await (await call(bucket, "GET", `${BASE}/a/b/c/abcdef`, { waitUntil: (promise) => pending.push(promise) })).arrayBuffer(); + expect(pending).toEqual([]); + expect(store.get(key)?.uploaded.getTime()).toBe(NOW_MS - 6 * DAY); + }); +}); + describe("compile-cache limits", () => { it("records its defaults and reads overrides from the Worker vars", () => { expect(resolveCompileCacheLimits({})).toEqual({ maxObjectBytes: 100 * 1024 * 1024, maxRepoBytes: 20 * 1024 * 1024 * 1024, + maxExecutionBytes: 10 * 1024 * 1024 * 1024, ttlDays: 14, }); expect( @@ -263,48 +419,85 @@ describe("compile-cache limits", () => { COMPILE_CACHE_MAX_REPO_BYTES: "garbage", COMPILE_CACHE_TTL_DAYS: "3", }), - ).toEqual({ maxObjectBytes: 10, maxRepoBytes: 20 * 1024 * 1024 * 1024, ttlDays: 3 }); + ).toEqual({ maxObjectBytes: 10, maxRepoBytes: 20 * 1024 * 1024 * 1024, maxExecutionBytes: 10 * 1024 * 1024 * 1024, ttlDays: 3 }); }); }); describe("sweepCompileCache", () => { const DAY = 24 * 3600 * 1000; - const limits = { maxObjectBytes: 100, maxRepoBytes: 100, ttlDays: 7 }; + const limits = { maxObjectBytes: 100, maxRepoBytes: 100, maxExecutionBytes: 100, ttlDays: 7 }; + const repoBytes = async (repo: string) => + (await ledgers.ledger(repo).reserve({ executionId: "probe", expiresAt: NOW + 1, bytes: 0, maxRepoBytes: 1e12, maxExecutionBytes: 1e12 })).repoBytes; + const sweep = (bucket: CompileCacheBucket, maxPages?: number) => + sweepCompileCache(bucket, limits, NOW_MS, { + ledger: (repo) => ledgers.ledger(repo), + ...(maxPages !== undefined ? { maxPages } : {}), + }); it("deletes expired entries, evicts the oldest-written to 90% of the ceiling, and records the size", async () => { const { bucket, store, seed } = makeBucket(2); - seed("compile-cache/owner/name/r/t/expired", 10, NOW_MS - 8 * DAY); - seed("compile-cache/owner/name/r/t/oldest", 40, NOW_MS - 3 * DAY); - seed("compile-cache/owner/name/r/t/older", 40, NOW_MS - 2 * DAY); - seed("compile-cache/owner/name/r/t/newest", 40, NOW_MS - 1 * DAY); - seed("compile-cache/other/repo/r/t/kept", 5, NOW_MS); + seed("compile-cache/owner/name/@base/r/t/expired", 10, NOW_MS - 8 * DAY); + seed("compile-cache/owner/name/@base/r/t/oldest", 40, NOW_MS - 3 * DAY); + seed("compile-cache/owner/name/@base/r/t/older", 40, NOW_MS - 2 * DAY); + seed("compile-cache/owner/name/@pr-7/r/t/newest", 40, NOW_MS - 1 * DAY); + seed("compile-cache/other/repo/@base/r/t/kept", 5, NOW_MS); - const result = await sweepCompileCache(bucket, limits, NOW_MS); + const result = await sweep(bucket); expect(result.truncated).toBe(false); expect(result.repos).toEqual([ - { repo: "other/repo", bytes: 5, objects: 1, expired: 0, evicted: 0 }, - { repo: "owner/name", bytes: 80, objects: 2, expired: 1, evicted: 1 }, + { repo: "other/repo", bytes: 5, objects: 1, expired: 0, evicted: 0, complete: true }, + { repo: "owner/name", bytes: 80, objects: 2, expired: 1, evicted: 1, complete: true }, ]); - expect([...store.keys()].filter((key) => key.startsWith("compile-cache/")).sort()).toEqual([ - "compile-cache/other/repo/r/t/kept", - "compile-cache/owner/name/r/t/newest", - "compile-cache/owner/name/r/t/older", + expect([...store.keys()].sort()).toEqual([ + "compile-cache/other/repo/@base/r/t/kept", + "compile-cache/owner/name/@base/r/t/older", + "compile-cache/owner/name/@pr-7/r/t/newest", ]); - expect(store.get(usageKey("owner/name"))?.customMetadata?.bytes).toBe("80"); + expect(await repoBytes("owner/name")).toBe(80); + expect(await repoBytes("other/repo")).toBe(5); }); - it("deletes the usage record of a prefix that no longer holds entries", async () => { - const { bucket, store } = makeBucket(); - await bucket.put(usageKey("gone/repo"), "", { customMetadata: { bytes: "999" } }); - await sweepCompileCache(bucket, limits, NOW_MS); - expect(store.has(usageKey("gone/repo"))).toBe(false); + it("resumes where the last tick stopped, so entries past one tick's listing still expire", async () => { + const { bucket, store, seed } = makeBucket(2); + for (const repo of ["a/one", "b/two", "c/three"]) { + seed(`compile-cache/${repo}/@base/r/t/1-expired`, 10, NOW_MS - 8 * DAY); + seed(`compile-cache/${repo}/@base/r/t/2-kept`, 10, NOW_MS); + seed(`compile-cache/${repo}/@base/r/t/3-kept`, 10, NOW_MS); + } + + const first = await sweep(bucket, 2); + expect(first.truncated).toBe(true); + expect(first.repos).toEqual([ + { repo: "a/one", bytes: 20, objects: 2, expired: 1, evicted: 0, complete: true }, + { repo: "b/two", bytes: 0, objects: 0, expired: 1, evicted: 0, complete: false }, + ]); + expect(store.has(SWEEP_STATE_KEY)).toBe(true); + + const second = await sweep(bucket, 3); + expect(second.truncated).toBe(false); + expect(second.repos.map((repo) => [repo.repo, repo.complete, repo.expired])).toEqual([ + ["b/two", false, 0], + ["c/three", true, 1], + ]); + expect([...store.keys()].filter((key) => key.includes("expired"))).toEqual([]); + expect(store.has(SWEEP_STATE_KEY)).toBe(false); + }); + + it("releases the bytes it deletes from a repository it could not list whole", async () => { + const { bucket, seed } = makeBucket(1); + await ledgers.ledger("a/one").reserve({ executionId: "x", expiresAt: NOW + 60, bytes: 50, maxRepoBytes: 100, maxExecutionBytes: 100 }); + seed("compile-cache/a/one/@base/r/t/1-expired", 30, NOW_MS - 8 * DAY); + seed("compile-cache/a/one/@base/r/t/2-kept", 20, NOW_MS); + const result = await sweep(bucket, 1); + expect(result.repos).toEqual([{ repo: "a/one", bytes: 0, objects: 0, expired: 1, evicted: 0, complete: false }]); + expect(await repoBytes("a/one")).toBe(20); }); }); describe("compileCacheOverrides", () => { const mint = { - wanted: true, + lane: "pr-7" as CompileCacheLane | undefined, repo: "owner/name", executionId: EXEC, createdAtMs: NOW_MS, @@ -314,10 +507,12 @@ describe("compileCacheOverrides", () => { }; it("hands the run the endpoint and a token valid until creation plus the ceiling", async () => { - const overrides = await compileCacheOverrides(mint); - expect(overrides?.["compile-cache.endpoint"]).toBe("https://dispatch.example/v1/compile-cache"); - const verdict = await verifyCompileCacheToken(SECRET, "owner/name", overrides!["compile-cache.token"]!, NOW); - expect(verdict).toEqual({ ok: true, scope: { repo: "owner/name", executionId: EXEC, expiresAt: NOW + 9900 } }); + const minted = await compileCacheOverrides(mint); + expect(minted?.expiresAt).toBe(NOW + 9900); + expect(minted?.lane).toBe("pr-7"); + expect(minted?.overrides["compile-cache.endpoint"]).toBe("https://dispatch.example/v1/compile-cache"); + const verdict = await verifyCompileCacheToken(SECRET, "owner/name", minted!.overrides["compile-cache.token"]!, NOW); + expect(verdict).toEqual({ ok: true, scope: { repo: "owner/name", executionId: EXEC, lane: "pr-7", expiresAt: NOW + 9900 } }); }); it("mints the same token on every invocation of one execution", async () => { @@ -325,7 +520,7 @@ describe("compileCacheOverrides", () => { }); it("mints nothing for a run that does not want the cache or a deploy that cannot serve it", async () => { - expect(await compileCacheOverrides({ ...mint, wanted: false })).toBeUndefined(); + expect(await compileCacheOverrides({ ...mint, lane: undefined })).toBeUndefined(); expect(await compileCacheOverrides({ ...mint, publicOrigin: undefined })).toBeUndefined(); expect(await compileCacheOverrides({ ...mint, secret: undefined })).toBeUndefined(); }); diff --git a/apps/dispatcher/src/routes/compile-cache.ts b/apps/dispatcher/src/routes/compile-cache.ts index 1a3b80e..caaae4a 100644 --- a/apps/dispatcher/src/routes/compile-cache.ts +++ b/apps/dispatcher/src/routes/compile-cache.ts @@ -11,28 +11,49 @@ // MKCOL create a parent collection // // plus HEAD. Collections are virtual: every path ending in `/` exists, so a -// PUT costs one PROPFIND and never a MKCOL. Entries live in `RUNS_STORAGE` -// under `compile-cache///`; the path below the repository -// is the client's `SCCACHE_WEBDAV_KEY_PREFIX` remainder and key. +// PUT costs one PROPFIND and never a MKCOL. // // Every request carries `Authorization: Bearer ` (compile-cache-token.ts) // minted for the repository the path names. Missing → 401; a token for another // repository, a forged or an expired one → 403; no `HMAC_SECRET` → 503. // +// Entries live in `RUNS_STORAGE` under `compile-cache///@/`, +// where `` is the client's `SCCACHE_WEBDAV_KEY_PREFIX` remainder and +// key, and `` comes from the token, never the request: +// +// base writes `@base`, reads `@base` +// pr- writes `@pr-`, reads `@pr-` and then `@base` +// read-only writes nothing (PUT → 403), reads `@base` +// +// so an entry a pull request writes reaches only that pull request. +// // Bounds, read from the Worker vars (wrangler.jsonc) with these defaults: -// COMPILE_CACHE_MAX_OBJECT_BYTES one entry; a larger PUT is 413 -// COMPILE_CACHE_MAX_REPO_BYTES one repository's prefix; a PUT while the -// last measured size plus the entry exceeds -// it is 507 -// COMPILE_CACHE_TTL_DAYS entry age at which the sweep deletes it -// The sweep (`sweepCompileCache`, on COMPILE_CACHE_SWEEP_CRON) measures each -// prefix, deletes entries past the TTL, then deletes the oldest-written -// entries until the prefix is at most 90% of its ceiling, and records the -// measured size the PUT admission reads. A prefix can therefore exceed its -// ceiling by the writes of one sweep interval. +// COMPILE_CACHE_MAX_OBJECT_BYTES one entry; a larger PUT is 413 +// COMPILE_CACHE_MAX_REPO_BYTES one repository, all lanes; a PUT the +// ledger cannot fit is 507 +// COMPILE_CACHE_MAX_EXECUTION_BYTES one execution's writes; past it, 507 +// COMPILE_CACHE_TTL_DAYS entry age at which the sweep deletes it +// Every PUT reserves its length in the repository's ledger +// (compile-cache-ledger.ts) before it writes, and a PUT after the execution +// ended is 403; a deploy without the `COMPILE_CACHE_LEDGER` binding refuses +// every PUT with 503 and still serves reads. +// +// A GET that serves an entry older than half the TTL rewrites it in place +// after the response, so an entry's age is its time since last use and the +// sweep's oldest-first eviction removes the least recently used. +// +// The sweep (`sweepCompileCache`, on COMPILE_CACHE_SWEEP_CRON) lists up to +// `maxPages` pages per tick and resumes from the stored cursor on the next +// tick. It deletes every expired entry it lists. For a repository listed +// whole within one tick it also deletes the oldest entries until the +// repository is at most 90% of its ceiling and replaces the ledger's count +// with the measured size; for one listed in part, it subtracts the bytes it +// deleted. // // Spec: specs/adr/0007-cost-efficient-execution.md § Shared compile cache. +import type { CompileCacheLane } from "@fractalboxdev/flare-dispatch-core/primitives"; +import type { CompileCacheLedgerApi } from "../compile-cache-ledger"; import { resolveCompileCacheSecret, verifyCompileCacheToken } from "../compile-cache-token"; import type { Env } from "../env"; @@ -40,18 +61,21 @@ import type { Env } from "../env"; export const COMPILE_CACHE_ROUTE = "/v1/compile-cache"; /** R2 prefix of the entries. */ export const COMPILE_CACHE_PREFIX = "compile-cache/"; -/** R2 prefix of the per-repository measured sizes the sweep records. */ -export const COMPILE_CACHE_USAGE_PREFIX = "compile-cache-usage/"; +/** R2 key of the sweep's resume point: the listing cursor and the repository it stopped in. */ +export const SWEEP_STATE_KEY = "compile-cache-sweep/state"; /** The sweep's cron (wrangler.jsonc `triggers.crons`). */ export const COMPILE_CACHE_SWEEP_CRON = "41 * * * *"; export const DEFAULT_MAX_OBJECT_BYTES = 100 * 1024 * 1024; export const DEFAULT_MAX_REPO_BYTES = 20 * 1024 * 1024 * 1024; +export const DEFAULT_MAX_EXECUTION_BYTES = 10 * 1024 * 1024 * 1024; export const DEFAULT_TTL_DAYS = 14; /** The sweep evicts down to this fraction of the ceiling, so the next writes have room. */ const EVICT_TO_RATIO = 0.9; -/** Listing pages one sweep reads (1000 entries each). */ +/** Listing pages one sweep tick reads (1000 entries each). */ const MAX_SWEEP_PAGES = 500; +/** A GET rewrites an entry older than this fraction of the TTL. */ +const REFRESH_AGE_RATIO = 0.5; const MAX_KEY_BYTES = 1024; const OWNER = /^[A-Za-z0-9][A-Za-z0-9-]{0,38}$/; @@ -61,6 +85,7 @@ const SEGMENT = /^[A-Za-z0-9._+-]{1,255}$/; export type CompileCacheLimits = { readonly maxObjectBytes: number; readonly maxRepoBytes: number; + readonly maxExecutionBytes: number; readonly ttlDays: number; }; @@ -71,18 +96,25 @@ const positive = (raw: string | undefined, fallback: number): number => { /** The configured bounds; an absent or malformed var takes its default. */ export const resolveCompileCacheLimits = ( - env: Pick, + env: Pick< + Env, + | "COMPILE_CACHE_MAX_OBJECT_BYTES" + | "COMPILE_CACHE_MAX_REPO_BYTES" + | "COMPILE_CACHE_MAX_EXECUTION_BYTES" + | "COMPILE_CACHE_TTL_DAYS" + >, ): CompileCacheLimits => ({ maxObjectBytes: positive(env.COMPILE_CACHE_MAX_OBJECT_BYTES, DEFAULT_MAX_OBJECT_BYTES), maxRepoBytes: positive(env.COMPILE_CACHE_MAX_REPO_BYTES, DEFAULT_MAX_REPO_BYTES), + maxExecutionBytes: positive(env.COMPILE_CACHE_MAX_EXECUTION_BYTES, DEFAULT_MAX_EXECUTION_BYTES), ttlDays: positive(env.COMPILE_CACHE_TTL_DAYS, DEFAULT_TTL_DAYS), }); -/** A request path resolved to its repository and R2 key. */ +/** A request path resolved to its repository and the key below it. */ export type CompileCachePath = { /** `owner/name`, lowercase. */ readonly repo: string; - /** The R2 key; ends in `/` for a collection. */ + /** The key below the repository, without a lane; ends in `/` for a collection. */ readonly key: string; readonly collection: boolean; }; @@ -110,13 +142,30 @@ export const parseCompileCachePath = (pathname: string): CompileCachePath | unde // `/` alone is the repository's root collection, never an entry. if (below.length === 0 && !collection) return undefined; const repo = `${owner}/${name}`.toLowerCase(); - const key = `${COMPILE_CACHE_PREFIX}${repo}/${below.join("/")}${below.length > 0 && collection ? "/" : ""}`; - if (new TextEncoder().encode(key).length > MAX_KEY_BYTES) return undefined; + const key = `${below.join("/")}${below.length > 0 && collection ? "/" : ""}`; + if (new TextEncoder().encode(laneKey(repo, "read-only", key)).length > MAX_KEY_BYTES) return undefined; return { repo, key, collection: collection || below.length === 0 }; }; -/** The R2 key of a repository's measured size. */ -export const usageKey = (repo: string): string => `${COMPILE_CACHE_USAGE_PREFIX}${repo.toLowerCase()}`; +/** The lane-qualified R2 key of `key` in `repo`; the longest lane name bounds the key length. */ +const laneKey = (repo: string, lane: string, key: string): string => + `${COMPILE_CACHE_PREFIX}${repo}/@${lane}/${key}`; + +/** The lanes a token reads, in order; the first holding the key answers. */ +const readLanes = (lane: CompileCacheLane): readonly string[] => + lane === "base" || lane === "read-only" ? ["base"] : [lane, "base"]; + +/** The ledger a deploy binds, by repository; `undefined` without the binding. */ +export type CompileCacheLedgers = (repo: string) => CompileCacheLedgerApi | undefined; + +/** The live ledger lookup over `COMPILE_CACHE_LEDGER`. */ +export const ledgersOf = + (env: Pick): CompileCacheLedgers => + (repo) => { + const namespace = env.COMPILE_CACHE_LEDGER; + if (namespace === undefined) return undefined; + return namespace.get(namespace.idFromName(repo.toLowerCase())) as unknown as CompileCacheLedgerApi; + }; /** The R2 surface the route and the sweep use. */ export type CompileCacheBucket = Pick; @@ -138,15 +187,38 @@ const xmlEscape = (value: string): string => /** The last-modified of a virtual collection: a fixed instant, since it never changes. */ const COLLECTION_MODIFIED = new Date(0).toUTCString(); -/** Ports the handler reads: the clock (epoch ms). */ -export type CompileCachePorts = { readonly now: () => number }; +/** + * Ports the handler reads: the clock (epoch ms), and for the rewrite of an + * aged entry, the request's `waitUntil` and a stream of known length (R2 + * stores a stream only when its length is known). Without `waitUntil` no + * entry is rewritten. + */ +export type CompileCachePorts = { + readonly now: () => number; + readonly waitUntil?: (promise: Promise) => void; + readonly fixedLength?: (stream: ReadableStream, length: number) => ReadableStream; +}; + +/** The Workers runtime's `FixedLengthStream`, or `undefined` outside it. */ +const workersFixedLength = (stream: ReadableStream, length: number): ReadableStream | undefined => { + const Fixed = (globalThis as { FixedLengthStream?: new (length: number) => TransformStream }).FixedLengthStream; + if (Fixed === undefined) return undefined; + const fixed = new Fixed(length); + stream.pipeTo(fixed.writable).catch(() => {}); + return fixed.readable; +}; /** Serve one compile-cache request. */ export const handleCompileCache = async ( request: Request, env: Pick< Env, - "HMAC_SECRET" | "COMPILE_CACHE_MAX_OBJECT_BYTES" | "COMPILE_CACHE_MAX_REPO_BYTES" | "COMPILE_CACHE_TTL_DAYS" + | "HMAC_SECRET" + | "COMPILE_CACHE_MAX_OBJECT_BYTES" + | "COMPILE_CACHE_MAX_REPO_BYTES" + | "COMPILE_CACHE_MAX_EXECUTION_BYTES" + | "COMPILE_CACHE_TTL_DAYS" + | "COMPILE_CACHE_LEDGER" > & { readonly RUNS_STORAGE: CompileCacheBucket }, ports: CompileCachePorts = { now: Date.now }, ): Promise => { @@ -159,11 +231,23 @@ export const handleCompileCache = async ( const auth = request.headers.get("authorization"); const bearer = auth === null ? null : /^Bearer\s+(\S+)$/i.exec(auth.trim()); if (bearer === null) return text(401, "missing bearer token", { "www-authenticate": 'Bearer realm="compile-cache"' }); - const verdict = await verifyCompileCacheToken(secret, path.repo, bearer[1]!, Math.floor(ports.now() / 1000)); + const nowSec = Math.floor(ports.now() / 1000); + const verdict = await verifyCompileCacheToken(secret, path.repo, bearer[1]!, nowSec); if (!verdict.ok) return text(403, `token ${verdict.reason}`); + const scope = verdict.scope; const bucket = env.RUNS_STORAGE; const limits = resolveCompileCacheLimits(env); + const lanes = readLanes(scope.lane); + /** The first object a read lane holds, with the R2 key it holds it under. */ + const find = async (read: (key: string) => Promise): Promise<{ key: string; object: T } | null> => { + for (const lane of lanes) { + const key = laneKey(path.repo, lane, path.key); + const object = await read(key); + if (object !== null) return { key, object }; + } + return null; + }; switch (request.method) { case "OPTIONS": return new Response(null, { status: 200, headers: { allow: ALLOW, dav: "1" } }); @@ -177,12 +261,28 @@ export const handleCompileCache = async ( "last-modified": object.uploaded.toUTCString(), }); if (request.method === "HEAD") { - const object = await bucket.head(path.key); - if (object === null) return text(404, "not found"); - return new Response(null, { status: 200, headers: entryHeaders(object) }); + const found = await find((key) => bucket.head(key)); + if (found === null) return text(404, "not found"); + return new Response(null, { status: 200, headers: entryHeaders(found.object) }); + } + const found = await find((key) => bucket.get(key)); + if (found === null) return text(404, "not found"); + const object = found.object; + const refreshBefore = ports.now() - limits.ttlDays * 24 * 3600 * 1000 * REFRESH_AGE_RATIO; + if (ports.waitUntil !== undefined && object.uploaded.getTime() < refreshBefore) { + const [served, copied] = object.body.tee(); + const sized = (ports.fixedLength ?? workersFixedLength)(copied, object.size); + if (sized === undefined) { + copied.cancel().catch(() => {}); + } else { + ports.waitUntil( + bucket + .put(found.key, sized, { httpMetadata: { contentType: "application/octet-stream" } }) + .catch(() => undefined), + ); + } + return new Response(served, { status: 200, headers: entryHeaders(object) }); } - const object = await bucket.get(path.key); - if (object === null) return text(404, "not found"); return new Response(object.body, { status: 200, headers: entryHeaders(object) }); } case "PUT": { @@ -192,11 +292,24 @@ export const handleCompileCache = async ( const length = Number(declared); if (length > limits.maxObjectBytes) return text(413, `entry of ${length} bytes exceeds the ${limits.maxObjectBytes}-byte bound`); - const usage = await bucket.head(usageKey(path.repo)); - const measured = Number(usage?.customMetadata?.bytes ?? "0"); - if (Number.isFinite(measured) && measured + length > limits.maxRepoBytes) - return text(507, `repository cache at ${measured} bytes of its ${limits.maxRepoBytes}-byte bound`); - const stored = await bucket.put(path.key, length === 0 || request.body === null ? "" : request.body, { + if (scope.lane === "read-only") return text(403, "token grants no writes"); + const ledger = ledgersOf(env)(path.repo); + if (ledger === undefined) return text(503, "compile cache writes unavailable: no ledger"); + const reserved = await ledger.reserve({ + executionId: scope.executionId, + expiresAt: scope.expiresAt, + bytes: length, + maxRepoBytes: limits.maxRepoBytes, + maxExecutionBytes: limits.maxExecutionBytes, + }); + if (!reserved.ok) { + if (reserved.reason === "ended") return text(403, "execution ended"); + return reserved.reason === "execution" + ? text(507, `execution's writes would pass its ${limits.maxExecutionBytes}-byte budget`) + : text(507, `repository cache at ${reserved.repoBytes} bytes of its ${limits.maxRepoBytes}-byte bound`); + } + const key = laneKey(path.repo, scope.lane, path.key); + const stored = await bucket.put(key, length === 0 || request.body === null ? "" : request.body, { httpMetadata: { contentType: "application/octet-stream" }, }); return new Response(null, { status: 201, headers: stored === null ? {} : { etag: stored.httpEtag } }); @@ -212,8 +325,9 @@ export const handleCompileCache = async ( href, `${COLLECTION_MODIFIED}`, ); - const object = await bucket.head(path.key); - if (object === null) return text(404, "not found"); + const found = await find((key) => bucket.head(key)); + if (found === null) return text(404, "not found"); + const object = found.object; return multistatus( href, `${object.uploaded.toUTCString()}${object.size}${xmlEscape(object.httpEtag)}`, @@ -227,18 +341,27 @@ export const handleCompileCache = async ( /** One repository's sweep outcome. */ export type CompileCacheSweepRepo = { readonly repo: string; + /** Bytes the repository holds after the sweep; for a partial listing, of the entries listed. */ readonly bytes: number; readonly objects: number; readonly expired: number; readonly evicted: number; + /** The tick listed the whole repository, so it evicted and measured. */ + readonly complete: boolean; }; export type CompileCacheSweepResult = { readonly repos: readonly CompileCacheSweepRepo[]; - /** The listing stopped at the page bound; sizes are lower bounds. */ + /** The tick stopped at its page bound; the next tick resumes from its cursor. */ readonly truncated: boolean; }; +export type CompileCacheSweepOptions = { + readonly ledger: CompileCacheLedgers; + /** Listing pages this tick reads; default 500. */ + readonly maxPages?: number; +}; + const repoOf = (key: string): string | undefined => { const parts = key.slice(COMPILE_CACHE_PREFIX.length).split("/"); return parts.length >= 3 ? `${parts[0]}/${parts[1]}` : undefined; @@ -248,74 +371,92 @@ const deleteKeys = async (bucket: CompileCacheBucket, keys: readonly string[]): for (let i = 0; i < keys.length; i += 1000) await bucket.delete(keys.slice(i, i + 1000)); }; +type Listed = { key: string; size: number; uploaded: number }; + /** - * Measure every repository prefix, delete entries older than the TTL, evict - * the oldest-written entries of a prefix over its ceiling down to 90% of it, - * and record each prefix's remaining size for the PUT admission. A usage - * record whose prefix holds no entries is deleted. + * One tick of the sweep. The listing is sorted by key, so each repository's + * entries are contiguous and only the current repository's are held. The tick + * starts from the stored cursor, deletes every expired entry it lists, and + * for each repository it lists whole evicts the oldest entries down to 90% of + * the ceiling and replaces the ledger's count with the remaining size. A + * repository the tick lists in part (the one it resumed in, the one it + * stopped in) has the bytes it deleted subtracted from its count instead. */ export const sweepCompileCache = async ( bucket: CompileCacheBucket, limits: CompileCacheLimits, nowMs: number, + options: CompileCacheSweepOptions, ): Promise => { - const byRepo = new Map(); - let cursor: string | undefined; - let truncated = false; - for (let page = 0; ; page++) { - if (page >= MAX_SWEEP_PAGES) { - truncated = true; - break; + const maxPages = options.maxPages ?? MAX_SWEEP_PAGES; + const nowSec = Math.floor(nowMs / 1000); + const cutoff = nowMs - limits.ttlDays * 24 * 3600 * 1000; + const target = Math.floor(limits.maxRepoBytes * EVICT_TO_RATIO); + const state = await bucket.head(SWEEP_STATE_KEY); + const resumedCursor = state?.customMetadata?.cursor; + const resumedRepo = state?.customMetadata?.repo; + + const repos: CompileCacheSweepRepo[] = []; + let current: { repo: string; complete: boolean; kept: Listed[]; expired: Listed[] } | undefined; + + const finish = async (complete: boolean): Promise => { + if (current === undefined) return; + const { repo, kept, expired } = current; + await deleteKeys(bucket, expired.map((entry) => entry.key)); + let bytes = kept.reduce((sum, entry) => sum + entry.size, 0); + const evicted: string[] = []; + if (complete) { + if (bytes > limits.maxRepoBytes) { + for (const entry of [...kept].sort((a, b) => a.uploaded - b.uploaded)) { + if (bytes <= target) break; + evicted.push(entry.key); + bytes -= entry.size; + } + await deleteKeys(bucket, evicted); + } + await options.ledger(repo)?.measure({ bytes, nowSec }); + } else { + const released = expired.reduce((sum, entry) => sum + entry.size, 0); + await options.ledger(repo)?.release({ bytes: released, nowSec }); } + repos.push({ repo, bytes, objects: kept.length - evicted.length, expired: expired.length, evicted: evicted.length, complete }); + current = undefined; + }; + + let cursor = resumedCursor; + let first = true; + let finished = false; + for (let page = 0; page < maxPages; page++) { const listed = await bucket.list({ prefix: COMPILE_CACHE_PREFIX, limit: 1000, ...(cursor !== undefined ? { cursor } : {}) }); for (const object of listed.objects) { const repo = repoOf(object.key); if (repo === undefined) continue; - const entries = byRepo.get(repo) ?? []; - entries.push({ key: object.key, size: object.size, uploaded: object.uploaded.getTime() }); - byRepo.set(repo, entries); + if (current?.repo !== repo) { + await finish(current?.complete ?? true); + // A repository the previous tick stopped in started before this tick's listing. + const complete = !(first && resumedCursor !== undefined && repo === resumedRepo); + current = { repo, complete, kept: [], expired: [] }; + first = false; + } + const entry = { key: object.key, size: object.size, uploaded: object.uploaded.getTime() }; + (entry.uploaded < cutoff ? current.expired : current.kept).push(entry); + } + if (!listed.truncated) { + finished = true; + break; } - if (!listed.truncated) break; cursor = listed.cursor; } - const cutoff = nowMs - limits.ttlDays * 24 * 3600 * 1000; - const target = Math.floor(limits.maxRepoBytes * EVICT_TO_RATIO); - const repos: CompileCacheSweepRepo[] = []; - for (const [repo, entries] of byRepo) { - const expired = entries.filter((entry) => entry.uploaded < cutoff); - const kept = entries.filter((entry) => entry.uploaded >= cutoff).sort((a, b) => a.uploaded - b.uploaded); - let bytes = kept.reduce((sum, entry) => sum + entry.size, 0); - const evicted: string[] = []; - if (bytes > limits.maxRepoBytes) { - for (const entry of kept) { - if (bytes <= target) break; - evicted.push(entry.key); - bytes -= entry.size; - } - } - await deleteKeys(bucket, [...expired.map((entry) => entry.key), ...evicted]); - const objects = kept.length - evicted.length; - await bucket.put(usageKey(repo), "", { - customMetadata: { bytes: String(bytes), objects: String(objects), measuredAt: new Date(nowMs).toISOString() }, + if (finished) { + await finish(current?.complete ?? true); + if (state !== null) await bucket.delete(SWEEP_STATE_KEY); + } else { + const stoppedIn = current?.repo; + await finish(false); + await bucket.put(SWEEP_STATE_KEY, "", { + customMetadata: { ...(cursor !== undefined ? { cursor } : {}), ...(stoppedIn !== undefined ? { repo: stoppedIn } : {}) }, }); - repos.push({ repo, bytes, objects, expired: expired.length, evicted: evicted.length }); - } - - if (!truncated) { - const stale: string[] = []; - let usageCursor: string | undefined; - do { - const listed = await bucket.list({ - prefix: COMPILE_CACHE_USAGE_PREFIX, - limit: 1000, - ...(usageCursor !== undefined ? { cursor: usageCursor } : {}), - }); - for (const object of listed.objects) - if (!byRepo.has(object.key.slice(COMPILE_CACHE_USAGE_PREFIX.length))) stale.push(object.key); - usageCursor = listed.truncated ? listed.cursor : undefined; - } while (usageCursor !== undefined); - await deleteKeys(bucket, stale); } - return { repos, truncated }; + return { repos, truncated: !finished }; }; diff --git a/apps/dispatcher/src/routes/scheduled.ts b/apps/dispatcher/src/routes/scheduled.ts index 3ba4ba1..21ef015 100644 --- a/apps/dispatcher/src/routes/scheduled.ts +++ b/apps/dispatcher/src/routes/scheduled.ts @@ -55,7 +55,7 @@ import { type UndeliveredRow, } from "../delivery-sweep"; import { toInstanceId } from "../instance-id"; -import { COMPILE_CACHE_SWEEP_CRON, resolveCompileCacheLimits, sweepCompileCache } from "./compile-cache"; +import { COMPILE_CACHE_SWEEP_CRON, ledgersOf, resolveCompileCacheLimits, sweepCompileCache } from "./compile-cache"; import { type ScheduleMatch, schedulesByCron } from "../registry"; import type { Env } from "../env"; @@ -194,12 +194,15 @@ export const liveCompileCacheSweep = (env: Env): RunDeliverySweep => async () => { try { - const result = await sweepCompileCache(env.RUNS_STORAGE, resolveCompileCacheLimits(env), Date.now()); + const result = await sweepCompileCache(env.RUNS_STORAGE, resolveCompileCacheLimits(env), Date.now(), { + ledger: ledgersOf(env), + }); for (const repo of result.repos) { console.info( - `[compile-cache-sweep] ${repo.repo} bytes=${repo.bytes} objects=${repo.objects} expired=${repo.expired} evicted=${repo.evicted}${result.truncated ? " (listing truncated)" : ""}`, + `[compile-cache-sweep] ${repo.repo} bytes=${repo.bytes} objects=${repo.objects} expired=${repo.expired} evicted=${repo.evicted}${repo.complete ? "" : " (listed in part)"}`, ); } + if (result.truncated) console.info("[compile-cache-sweep] page bound reached; the next tick resumes from its cursor"); } catch (cause) { console.error( `[compile-cache-sweep] tick failed: ${cause instanceof Error ? cause.message : String(cause)}`, diff --git a/apps/dispatcher/src/run-end.test.ts b/apps/dispatcher/src/run-end.test.ts new file mode 100644 index 0000000..2c3d882 --- /dev/null +++ b/apps/dispatcher/src/run-end.test.ts @@ -0,0 +1,70 @@ +// What the workflow revokes when an execution's run body ends: the self-heal +// budget of an agent run, and the compile-cache writes of a run that holds a +// writable grant. + +import { describe, expect, it, vi } from "vitest"; +import { revokeAtRunEnd } from "./run-end"; + +const EXPIRES = 1_800_003_600; + +const ports = () => { + const kill = vi.fn(async () => {}); + const end = vi.fn(async (_: { executionId: string; expiresAt: number }) => {}); + return { + kill, + end, + agentBudget: () => ({ kill }), + ledger: (_repo: string) => ({ end }), + }; +}; + +describe("revokeAtRunEnd", () => { + it("kills the agent budget of a self-heal run and nothing else", async () => { + const p = ports(); + await revokeAtRunEnd({ engineInterrupted: false, agentBudgetArmed: true, executionId: "e", repo: "o/n" }, p); + expect(p.kill).toHaveBeenCalledTimes(1); + expect(p.end).not.toHaveBeenCalled(); + }); + + it("ends a compile-cache run's writes without touching an agent budget it never armed", async () => { + const p = ports(); + await revokeAtRunEnd( + { engineInterrupted: false, agentBudgetArmed: false, executionId: "e", repo: "O/N", compileCache: { lane: "pr-7", expiresAt: EXPIRES } }, + p, + ); + expect(p.kill).not.toHaveBeenCalled(); + expect(p.end).toHaveBeenCalledWith({ executionId: "e", expiresAt: EXPIRES }); + }); + + it("has nothing to end for a read-only grant", async () => { + const p = ports(); + await revokeAtRunEnd( + { engineInterrupted: false, agentBudgetArmed: false, executionId: "e", repo: "o/n", compileCache: { lane: "read-only", expiresAt: EXPIRES } }, + p, + ); + expect(p.end).not.toHaveBeenCalled(); + }); + + it("revokes nothing when the engine interrupted the run, which resumes on replay", async () => { + const p = ports(); + await revokeAtRunEnd( + { engineInterrupted: true, agentBudgetArmed: true, executionId: "e", repo: "o/n", compileCache: { lane: "base", expiresAt: EXPIRES } }, + p, + ); + expect(p.kill).not.toHaveBeenCalled(); + expect(p.end).not.toHaveBeenCalled(); + }); + + it("swallows a failed revocation", async () => { + const failing = { + agentBudget: () => ({ kill: async () => Promise.reject(new Error("down")) }), + ledger: () => ({ end: async () => Promise.reject(new Error("down")) }), + }; + await expect( + revokeAtRunEnd( + { engineInterrupted: false, agentBudgetArmed: true, executionId: "e", repo: "o/n", compileCache: { lane: "base", expiresAt: EXPIRES } }, + failing, + ), + ).resolves.toBeUndefined(); + }); +}); diff --git a/apps/dispatcher/src/run-end.ts b/apps/dispatcher/src/run-end.ts new file mode 100644 index 0000000..77163f8 --- /dev/null +++ b/apps/dispatcher/src/run-end.ts @@ -0,0 +1,44 @@ +// FlareDispatch Dispatcher — capabilities the workflow revokes when an +// execution's run body ends. +// +// - The self-heal AgentBudget of an agent run: its capability token is +// deterministic and non-expiring, so `kill` stops the post-run tail +// (specs/08-self-healing.md § 6.3). Only a run that armed the budget calls +// it. +// - The compile-cache writes of a run holding a writable lane: `end` in the +// repository's ledger refuses the token's further PUTs, though the token +// stays valid for reads until it expires. +// +// An engine interruption revokes nothing: the Workflow replays the run, and +// the replay needs both. Every revocation is best-effort; a failure leaves the +// budget's hard cap and the token's expiry as the bound. + +import type { CompileCacheLane } from "@fractalboxdev/flare-dispatch-core/primitives"; + +export type RunEnd = { + readonly engineInterrupted: boolean; + /** The run armed the AgentBudget Durable Object at start. */ + readonly agentBudgetArmed: boolean; + readonly executionId: string; + readonly repo: string; + /** The compile-cache grant the run was minted, if any. */ + readonly compileCache?: { readonly lane: CompileCacheLane; readonly expiresAt: number }; +}; + +export type RunEndPorts = { + readonly agentBudget: (executionId: string) => { kill(): Promise } | undefined; + readonly ledger: ( + repo: string, + ) => { end(request: { readonly executionId: string; readonly expiresAt: number }): Promise } | undefined; +}; + +export const revokeAtRunEnd = async (end: RunEnd, ports: RunEndPorts): Promise => { + if (end.engineInterrupted) return; + if (end.agentBudgetArmed) await ports.agentBudget(end.executionId)?.kill().catch(() => {}); + if (end.compileCache !== undefined && end.compileCache.lane !== "read-only") { + await ports + .ledger(end.repo.toLowerCase()) + ?.end({ executionId: end.executionId, expiresAt: end.compileCache.expiresAt }) + .catch(() => {}); + } +}; diff --git a/apps/dispatcher/src/workflow.ts b/apps/dispatcher/src/workflow.ts index 2ed121d..65f36b0 100644 --- a/apps/dispatcher/src/workflow.ts +++ b/apps/dispatcher/src/workflow.ts @@ -131,6 +131,8 @@ import { logLinksSuffix, startedSummary } from "./summary-links"; import { resolveMailboxLinkSecret, signMailboxToken } from "./mailbox-token"; import { resolveAgentProxySecret, signAgentToken } from "./agent-token"; import { compileCacheOverrides, resolveCompileCacheSecret } from "./compile-cache-token"; +import { ledgersOf } from "./routes/compile-cache"; +import { revokeAtRunEnd } from "./run-end"; import { deliverSlackVerdict, emitSlackNotice, @@ -539,6 +541,7 @@ export class RunWorkflow extends WorkflowEntrypoint { // is a deterministic HMAC of the execution id, and `AgentBudget.init` ignores // a refill once spending has begun. let configOverrides: Record | undefined; + let agentBudgetArmed = false; if ( run.sandboxImage === "agent" && this.env.AGENT_BUDGET !== undefined && @@ -567,20 +570,21 @@ export class RunWorkflow extends WorkflowEntrypoint { "self-heal.proxy-url": proxyUrl, "self-heal.agent-token": token, }; + agentBudgetArmed = true; } } // Shared compile cache (routes/compile-cache.ts): a run whose definition - // asks for it gets the cache endpoint and a token scoped to this - // repository and execution. The token derives from the instance's + // names a lane gets the cache endpoint and a token scoped to this + // repository, execution and lane. The token derives from the instance's // creation time, so every invocation mints the same one and a check // command's fingerprint, which covers its env, holds across replays. const cacheQueueAgeSec = typeof run.limits.admissionMaxQueueAgeSec === "function" ? run.limits.admissionMaxQueueAgeSec(input) : (run.limits.admissionMaxQueueAgeSec ?? ADMISSION_MAX_QUEUE_AGE_MS / 1000); - const cacheOverrides = await compileCacheOverrides({ - wanted: run.compileCache?.(input) === true, + const cacheGrant = await compileCacheOverrides({ + lane: run.compileCache?.(input), repo: payload.github.repo, executionId: payload.executionId, createdAtMs: event.timestamp.getTime(), @@ -588,7 +592,7 @@ export class RunWorkflow extends WorkflowEntrypoint { publicOrigin, secret: resolveCompileCacheSecret(this.env), }); - if (cacheOverrides !== undefined) configOverrides = { ...configOverrides, ...cacheOverrides }; + if (cacheGrant !== undefined) configOverrides = { ...configOverrides, ...cacheGrant.overrides }; // Per-repo container transport pin. It outranks the Worker-wide // `SANDBOX_TRANSPORT` in either direction, and with neither set containers @@ -1729,19 +1733,24 @@ export class RunWorkflow extends WorkflowEntrypoint { ), ); } finally { - // Self-heal: mark the AgentBudget no longer live once the run ends, so the - // (deterministic, non-expiring) capability token can't drain the remaining - // budget after the execution finishes. Best-effort — the hard token cap - // already bounds spend; this revokes the post-run tail. specs/08 § 6.3. - if ( - !engineInterrupted && - configOverrides !== undefined && - this.env.AGENT_BUDGET !== undefined - ) { - await this.env.AGENT_BUDGET.get(this.env.AGENT_BUDGET.idFromName(payload.executionId)) - .kill() - .catch(() => {}); - } + // Revoke what the run held (run-end.ts): the self-heal AgentBudget of a + // run that armed it, so its non-expiring token cannot drain the budget + // after the execution (specs/08 § 6.3), and the compile-cache writes of + // a writable lane, so the token stops writing before it expires. + const agentBudget = this.env.AGENT_BUDGET; + await revokeAtRunEnd( + { + engineInterrupted, + agentBudgetArmed, + executionId: payload.executionId, + repo: payload.github.repo, + ...(cacheGrant !== undefined ? { compileCache: { lane: cacheGrant.lane, expiresAt: cacheGrant.expiresAt } } : {}), + }, + { + agentBudget: (executionId) => agentBudget?.get(agentBudget.idFromName(executionId)), + ledger: (repo) => ledgersOf(this.env)(repo), + }, + ); } } } diff --git a/infra/overlay-limits.mjs b/infra/overlay-limits.mjs index dacc821..537bcf5 100644 --- a/infra/overlay-limits.mjs +++ b/infra/overlay-limits.mjs @@ -1,10 +1,15 @@ -// Merge the committed wrangler.jsonc's `limits`, and each container's -// `rollout_*` settings, into the operator overlay. +// Merge the committed wrangler.jsonc's `limits`, each container's `rollout_*` +// settings, and its Durable Object bindings and migrations into the operator +// overlay. // // The deploy writes the overlay over the committed file, so a setting with no // account-specific value reaches production only through this merge. A value // the overlay sets wins; one it omits comes from the template. Rollout // settings join by `class_name`; a container the template lacks is unchanged. +// Durable Objects merge only when the overlay's migration tags are a strict +// prefix of the template's: the overlay then takes the template's migrations +// and every template binding it lacks. An overlay whose migrations diverge +// keeps its own Durable Objects unchanged. // // Usage: node infra/overlay-limits.mjs // rewrites in place as JSON. @@ -36,6 +41,20 @@ export const mergeOverlayLimits = (templateText, overlayText) => { return { ...container, ...rollout, ...container }; }); } + const templateMigrations = Array.isArray(template.migrations) ? template.migrations : []; + const overlayMigrations = Array.isArray(overlay.migrations) ? overlay.migrations : undefined; + if ( + overlayMigrations !== undefined && + overlayMigrations.length < templateMigrations.length && + overlayMigrations.every((migration, i) => migration?.tag === templateMigrations[i]?.tag) + ) { + overlay.migrations = templateMigrations; + const own = overlay.durable_objects?.bindings ?? []; + const added = (template.durable_objects?.bindings ?? []).filter( + (binding) => !own.some((existing) => existing.name === binding.name), + ); + overlay.durable_objects = { ...overlay.durable_objects, bindings: [...own, ...added] }; + } return `${JSON.stringify(overlay, null, 2)}\n`; }; @@ -49,6 +68,7 @@ if (process.argv[1] === fileURLToPath(import.meta.url)) { writeFileSync(overlayPath, merged); const config = JSON.parse(merged); console.log(`limits: ${JSON.stringify(config.limits ?? null)}`); + console.log(`durable objects: ${(config.durable_objects?.bindings ?? []).map((binding) => binding.class_name).join(", ")}`); for (const container of config.containers ?? []) { const rollout = Object.entries(container).filter(([key]) => key.startsWith("rollout_")); if (rollout.length > 0) console.log(`${container.class_name}: ${JSON.stringify(Object.fromEntries(rollout))}`); diff --git a/packages/core/src/define-run.ts b/packages/core/src/define-run.ts index 111375f..efe6c72 100644 --- a/packages/core/src/define-run.ts +++ b/packages/core/src/define-run.ts @@ -11,6 +11,7 @@ import type { RunContext } from "./context"; import type { RunError } from "./errors"; import type { AdmissionClass } from "./run-admission"; import type { WritebackSpec } from "./writeback"; +import type { CompileCacheLane } from "./primitives/compile-cache"; export type RunLimits = { readonly maxDurationSec: number; @@ -221,16 +222,18 @@ export type RunSpec = { */ readonly joinsChildren?: boolean; /** - * Whether this execution compiles through the shared compile cache. When it - * answers `true`, the dispatcher mints a token scoped to the execution's - * repository and execution, valid until the execution's queue age plus its - * `maxDurationSec` from creation, and injects it with the cache endpoint - * under `COMPILE_CACHE_TOKEN_KEY` / `COMPILE_CACHE_ENDPOINT_KEY` - * (primitives/compile-cache.ts). A deploy without a public origin or - * `HMAC_SECRET` injects neither, and the run compiles without the cache. + * The compile-cache lane this execution writes, or `undefined` when it + * compiles without the cache. For a lane, the dispatcher mints a token bound + * to the execution's repository, execution and lane, valid until the + * execution's queue age plus its `maxDurationSec` from creation, and injects + * it with the cache endpoint under `COMPILE_CACHE_TOKEN_KEY` / + * `COMPILE_CACHE_ENDPOINT_KEY` (primitives/compile-cache.ts). Only a + * default-branch execution answers `base`; a pull request's answers its own + * `pr-`. A deploy without a public origin or `HMAC_SECRET` injects + * neither, and the run compiles without the cache. */ - readonly compileCache?: (input: I) => boolean; - readonly run:(input: I) => Effect.Effect; + readonly compileCache?: (input: I) => CompileCacheLane | undefined; + readonly run: (input: I) => Effect.Effect; }; /** diff --git a/packages/core/src/primitives/compile-cache.test.ts b/packages/core/src/primitives/compile-cache.test.ts index 315ed3c..bd73ece 100644 --- a/packages/core/src/primitives/compile-cache.test.ts +++ b/packages/core/src/primitives/compile-cache.test.ts @@ -6,7 +6,7 @@ import { chmodSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "nod import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { compileCacheCommand, compileCacheEnv } from "./compile-cache"; +import { compileCacheCommand, compileCacheEnv, isCompileCacheLane } from "./compile-cache"; const dirs: string[] = []; afterEach(() => { @@ -91,6 +91,14 @@ describe("compileCacheCommand", () => { expect(box.log()).toBe(""); }); + it("runs a command that ends in a comment", () => { + const box = sandbox({ sccache: { startExit: 0 } }); + const result = box.run("echo ran # trailing comment"); + expect(result.status).toBe(0); + expect(result.stdout).toContain("ran"); + expect(result.stdout).toContain("Compile requests 12"); + }); + it("refuses a repository outside the owner/name charset", () => { expect(() => compileCacheCommand("true", 'owner/name"; rm -rf /')).toThrow(/invalid repo/); }); @@ -102,3 +110,14 @@ describe("compileCacheCommand", () => { }); }); }); + +describe("isCompileCacheLane", () => { + it("accepts the default-branch lane, a pull request's lane and the read-only lane", () => { + for (const lane of ["base", "read-only", "pr-1", "pr-4242"]) expect(isCompileCacheLane(lane)).toBe(true); + }); + + it("refuses anything else", () => { + for (const lane of ["", "pr-", "pr-0", "pr-01", "pr-1/x", "Base", "main", "../base", "pr-12345678901"]) + expect(isCompileCacheLane(lane)).toBe(false); + }); +}); diff --git a/packages/core/src/primitives/compile-cache.ts b/packages/core/src/primitives/compile-cache.ts index a2e0153..4f72f9d 100644 --- a/packages/core/src/primitives/compile-cache.ts +++ b/packages/core/src/primitives/compile-cache.ts @@ -3,7 +3,7 @@ // A run whose definition declares `compileCache` gets two config values the // dispatcher injects when the run starts (apps/dispatcher/src/workflow.ts): // the WebDAV endpoint of the dispatcher's compile-cache route and a token -// scoped to the run's repository and execution. `compileCacheCommand` wraps a +// scoped to the run's repository, execution and lane. `compileCacheCommand` wraps a // shell command so every `rustc` it invokes goes through sccache against that // endpoint, and prints sccache's statistics into the command's log when the // command ends. @@ -24,6 +24,23 @@ export const COMPILE_CACHE_ENDPOINT_KEY = "compile-cache.endpoint"; /** Config key carrying the run's compile-cache token. */ export const COMPILE_CACHE_TOKEN_KEY = "compile-cache.token"; +/** + * Where an execution's compile-cache writes land, bound into its token: + * + * - `base` the default branch's lane: reads and writes it. + * - `pr-` pull request `n`'s lane: writes it, reads it and then `base`. + * - `read-only` reads `base` and writes nothing. + * + * No execution reads another pull request's lane, so an entry a pull request + * writes reaches only that pull request's later executions. + */ +export type CompileCacheLane = "base" | "read-only" | `pr-${number}`; + +const LANE = /^(base|read-only|pr-[1-9][0-9]{0,9})$/; + +/** Whether `value` names a lane. */ +export const isCompileCacheLane = (value: string): value is CompileCacheLane => LANE.test(value); + /** The environment the wrapped command needs from the run. */ export type CompileCacheAccess = { readonly endpoint: string; @@ -55,6 +72,8 @@ export const compileCacheEnv = (access: CompileCacheAccess): Record { if (!REPO_SLUG.test(repo)) throw new Error(`compileCacheCommand: invalid repo ${JSON.stringify(repo)}`); @@ -77,7 +96,9 @@ export const compileCacheCommand = (command: string, repo: string): string => { "else", " echo 'compile cache: sccache, rustc or cache access unavailable; compiling locally' >&2", "fi", - `( ${command} )`, + "(", + command, + ")", "fd_cc_status=$?", "if [ \"$fd_cc_on\" = 1 ]; then", " echo '--- sccache --show-stats ---'", diff --git a/packages/core/src/primitives/index.ts b/packages/core/src/primitives/index.ts index 944ec36..8a5cb66 100644 --- a/packages/core/src/primitives/index.ts +++ b/packages/core/src/primitives/index.ts @@ -39,7 +39,9 @@ export { compileCacheEnv, COMPILE_CACHE_ENDPOINT_KEY, COMPILE_CACHE_TOKEN_KEY, + isCompileCacheLane, type CompileCacheAccess, + type CompileCacheLane, } from "./compile-cache"; export { sharded, type Shard } from "./sharded"; export { fanOut, type FanOutShard } from "./fan-out"; diff --git a/runs/check.test.ts b/runs/check.test.ts index 443dae2..9b84d08 100644 --- a/runs/check.test.ts +++ b/runs/check.test.ts @@ -565,7 +565,7 @@ describe("check", () => { expect(exec?.command).toContain("CARGO_INCREMENTAL=0"); expect(exec?.command).toContain("SCCACHE_IGNORE_SERVER_IO_ERROR=1"); expect(exec?.command).toContain("sccache --show-stats"); - expect(exec?.command).toContain(`( ${CHECK_CMD} )`); + expect(exec?.command).toContain(`(\n${CHECK_CMD}\n)`); expect(exec?.stdout ?? "").not.toContain(TOKEN); }).pipe(Effect.provide(layer)); }); @@ -592,9 +592,17 @@ describe("check", () => { }).pipe(Effect.provide(layer)); }); - it("asks the dispatcher for a token only when the input asks for the cache", () => { - expect(check.compileCache?.({ ...baseInput, compileCache: true } as never)).toBe(true); - expect(check.compileCache?.(baseInput as never)).toBe(false); + it("asks the dispatcher for a token in the input's lane, read-only when it names none", () => { + expect(check.compileCache?.({ ...baseInput, compileCache: true, compileCacheLane: "base" } as never)).toBe("base"); + expect(check.compileCache?.({ ...baseInput, compileCache: true, compileCacheLane: "pr-7" } as never)).toBe("pr-7"); + expect(check.compileCache?.({ ...baseInput, compileCache: true } as never)).toBe("read-only"); + expect(check.compileCache?.(baseInput as never)).toBeUndefined(); + }); + + it("refuses a lane outside the lane grammar", () => { + expect(() => + Schema.decodeUnknownSync(check.inputs)({ ...baseInput, compileCache: true, compileCacheLane: "pr-1/../base" }), + ).toThrow(); }); }); diff --git a/runs/check.ts b/runs/check.ts index 27ec2b1..d8d7fb3 100644 --- a/runs/check.ts +++ b/runs/check.ts @@ -61,6 +61,7 @@ import { COMPILE_CACHE_ENDPOINT_KEY, COMPILE_CACHE_TOKEN_KEY, durableExec, + isCompileCacheLane, ensureWorkspace, isRestorableEntry, loadSecrets, @@ -135,6 +136,12 @@ const CheckInput = Schema.Struct({ * injected token the command runs unwrapped. */ compileCache: Schema.optional(Schema.Boolean), + /** + * The compile-cache lane the token writes (primitives/compile-cache.ts + * `CompileCacheLane`): `base` for a default-branch execution, `pr-` for + * pull request `n`'s. Absent, the token reads `base` and writes nothing. + */ + compileCacheLane: Schema.optional(Schema.String.pipe(Schema.filter(isCompileCacheLane))), /** * Non-sensitive env only. Dispatch inputs are persisted (executions row + * Workflow params) — never put credentials here. Same contract as @@ -252,8 +259,10 @@ export const check = defineRun({ // unlabelled dispatch resolves the same key, names the same check-run, and // gets the same instance id as under 1.0.0. // 1.2.0 — additive: `restore`. A dispatch without it runs as under 1.1.0. + // 1.4.0 — additive: `compileCacheLane`. A dispatch without it reads the + // compile cache and writes nothing. // 1.3.0 — additive: `compileCache`. A dispatch without it runs as under 1.2.0. - version: "1.3.0", + version: "1.4.0", inputs: CheckInput, outputs: CheckOutput, @@ -296,7 +305,12 @@ export const check = defineRun({ optIn: ({ input }) => input.command !== undefined ? [] : commandKeys(input.repo, input.checkLabel), - compileCache: (input) => input.compileCache === true, + compileCache: (input) => + input.compileCache === true + ? input.compileCacheLane !== undefined && isCompileCacheLane(input.compileCacheLane) + ? input.compileCacheLane + : "read-only" + : undefined, limits: { // Run-body wall-time ceiling, enforced by the dispatcher: one full 2700 s diff --git a/runs/contextful-gate.test.ts b/runs/contextful-gate.test.ts index 4851ee0..0ca60b6 100644 --- a/runs/contextful-gate.test.ts +++ b/runs/contextful-gate.test.ts @@ -16,6 +16,7 @@ import { contextfulGate, lightLabelsKey, mergeStages, + compileCacheLane, NOT_AFFECTED_SUMMARY, parseAffected, parseLightLabels, @@ -322,6 +323,7 @@ describe("contextful-gate", () => { checkLabel: stage, install: true, compileCache: true, + compileCacheLane: "read-only", restore, command: `if [ -x ${BIN} ] && ${BIN} --help >/dev/null 2>&1; then exec ${BIN} gate --predecessors --stage ${stage} --base ${BASE_SHA}; else exec ${CARGO_RUN} gate --predecessors --stage ${stage} --base ${BASE_SHA}; fi`, }), @@ -999,6 +1001,12 @@ describe("contextful-gate", () => { expect(tierParts(parts, ["absent"], "heavy")).toEqual(parts); }); + it("writes the default branch's compile-cache lane only from a default-branch push", () => { + expect(compileCacheLane(heavyInput)).toBe("base"); + expect(compileCacheLane({ ...input, pr: 42 })).toBe("pr-42"); + expect(compileCacheLane(input)).toBe("read-only"); + }); + it("unset: a pull-request gate runs every part and the heavy tier none", () => { expect(tierParts(parts, undefined, "pull-request")).toBe(parts); expect(tierParts(parts, undefined, "heavy")).toEqual([]); @@ -1047,6 +1055,8 @@ describe("contextful-gate", () => { deferred: ["pins"], }); expect(spawnedLabels(handles)).toEqual(["workspace.compile", "formal"]); + for (const spawn of handles.childRuns.spawned) + expect(spawn.input).toEqual(expect.objectContaining({ compileCache: true, compileCacheLane: "base" })); expect(handles.github.commitPullRequestsCalls).toHaveLength(0); expect(handles.github.pullRequestCalls).toHaveLength(0); }).pipe(Effect.provide(layer)); diff --git a/runs/contextful-gate.ts b/runs/contextful-gate.ts index 4731149..ee7264b 100644 --- a/runs/contextful-gate.ts +++ b/runs/contextful-gate.ts @@ -22,6 +22,7 @@ import { prebuiltCommand, waitForAnyChildDurably, workspace, + type CompileCacheLane, type PrebuiltBinary, } from "@fractalboxdev/flare-dispatch-core/primitives"; import { check } from "./check"; @@ -219,6 +220,13 @@ const Input = Schema.Struct({ type InputType = Schema.Schema.Type; +/** + * The compile-cache lane a gate's parts write: `base` from a default-branch + * push, the pull request's own lane from a pull-request gate, and none + * (read-only) from a gate that names no pull request. + */ +export const compileCacheLane = (input: InputType): CompileCacheLane => (input.tier === "heavy" ? "base" : input.pr !== undefined ? `pr-${input.pr}` : "read-only"); + /** * A reopen ranks this far above its `updated_at`, past the close marker's * 1 ms (apps/dispatcher/src/pr-close.ts): GitHub reports `updated_at` to the @@ -336,11 +344,13 @@ export const planStages = ( export const contextfulGate = defineRun({ name: "contextful-gate", + // 1.5.0 — a part writes the compile cache's default-branch lane only from a + // default-branch push, and its own pull request's lane otherwise. // 1.4.0 — every part compiles through the shared compile cache. // 1.3.0 — two-tier gate: `gate.lightLabels:` splits light parts (pull // requests) from heavy parts (pushes to the default branch). // 1.2.2 — a reopen ranks above the close marker of its own second. - version: "1.4.0", + version: "1.5.0", inputs: Input, outputs: Output, triggers: [ @@ -638,6 +648,7 @@ export const contextfulGate = defineRun({ // cache (sccache over the dispatcher's compile-cache route) // replaces each part's cold compile with cache reads. compileCache: true, + compileCacheLane: compileCacheLane(input), ...(prebuilt === null ? {} : { restore: [{ key: prebuilt.key, base: prebuilt.base, paths: prebuilt.paths }] }), diff --git a/specs/adr/0007-cost-efficient-execution.md b/specs/adr/0007-cost-efficient-execution.md index 686bae2..620c2eb 100644 --- a/specs/adr/0007-cost-efficient-execution.md +++ b/specs/adr/0007-cost-efficient-execution.md @@ -73,20 +73,30 @@ so a budget buys more checks. ### Shared compile cache -- A run whose definition answers `compileCache` with `true` compiles Rust +- A run whose definition answers `compileCache` with a lane compiles Rust through sccache. Every contextful gate part does. The dispatcher's `/v1/compile-cache` route serves sccache's WebDAV backend (GET, HEAD, PUT, `Depth: 0` PROPFIND, MKCOL) from `RUNS_STORAGE` under - `compile-cache///`; the client's key prefix below that is - `//`. + `compile-cache///@/`; the client's key prefix below that + is `//`. - The dispatcher mints one token per execution: an HMAC under a key derived - from `HMAC_SECRET`, bound to the repository and execution, expiring at the - instance's creation plus its queue age and `maxDurationSec`. A token for + from `HMAC_SECRET`, bound to the repository, execution and lane, expiring at + the instance's creation plus its queue age and `maxDurationSec`. A token for another repository, a forged one and an expired one are refused. -- One entry is at most `COMPILE_CACHE_MAX_OBJECT_BYTES`, one repository prefix - at most `COMPILE_CACHE_MAX_REPO_BYTES` as of the last hourly sweep, and an - entry older than `COMPILE_CACHE_TTL_DAYS` is deleted. The sweep evicts the - oldest-written entries down to 90% of the ceiling. +- The lane decides writes and reads. `base` belongs to default-branch + executions and reads and writes `@base`. `pr-` writes `@pr-` and reads + it, then `@base`. `read-only` reads `@base` and writes nothing. No execution + reads another pull request's lane. +- One entry is at most `COMPILE_CACHE_MAX_OBJECT_BYTES`. Every PUT reserves its + length in the repository's ledger, a Durable Object, against + `COMPILE_CACHE_MAX_REPO_BYTES` for the repository and + `COMPILE_CACHE_MAX_EXECUTION_BYTES` for the execution. When the run ends the + workflow ends the execution's record, and its token writes nothing more. +- The hourly sweep lists up to 500 pages per tick and resumes from its stored + cursor. It deletes entries older than `COMPILE_CACHE_TTL_DAYS`. For a + repository listed whole within a tick, it evicts the oldest entries down to + 90% of the ceiling and sets the ledger to the measured size. A GET of an + entry older than half the TTL rewrites it, so age counts from last use. - A cache outage never fails a part: the wrapper starts the sccache server first and compiles without `RUSTC_WRAPPER` when it does not start; a started server reads an unreachable entry as a miss. @@ -106,6 +116,13 @@ compiler cache turns all but the first compile of each crate into a read. sccache's WebDAV backend needs no cloud credentials in the container, so the Worker's R2 binding stays the only storage authority. +Pull-request code runs with its token in its environment, and sccache trusts +whatever an entry holds. A shared writable prefix would let one pull request +plant an artifact that another pull request's gate, or a default-branch gate, +links and runs. Lanes confine each pull request's writes to itself. A ledger +counted at admission keeps the byte bound against a client that ignores +sccache's write pattern, where a bound measured hourly would not. + Building a C++ engine from source in every part buys nothing a prebuilt library does not, and only the consumer can decide whether CI may link it dynamically while release builds stay static. @@ -117,9 +134,15 @@ dynamically while release builds stay static. - A broken restore still costs a cold install per part until the transport is fixed, but no longer a repeated pack and upload. - The lean image grows by the prebuilt libraries' size. -- Pull-request parts write to the repository's shared prefix, so a - same-repository pull request can place an entry a later default-branch - part reads. +- A pull request reuses entries across its own pushes and the default + branch's entries. It reuses nothing another pull request compiled. Only + default-branch executions write `@base`. A gate whose default-branch tier + runs other parts than its pull-request tier therefore populates `@base` only + for the compiles those parts share. +- A pull request's lane holds its entries until they expire, after the pull + request closes too. +- A deploy whose `wrangler.jsonc` lacks the `COMPILE_CACHE_LEDGER` binding + serves compile-cache reads and refuses every write. ## Revisit triggers diff --git a/specs/adr/README.md b/specs/adr/README.md index 828d01f..d23e3e4 100644 --- a/specs/adr/README.md +++ b/specs/adr/README.md @@ -17,6 +17,6 @@ interfaces or assume they exist. | [0004](./0004-serialized-run-groups.md) | proposed | A run may declare a serialization group: one execution in flight, newest waiter wins, never cancel the running one; state in D1 | | [0005](./0005-superseded-run-groups.md) | proposed | A run may declare a supersession group: the highest-ranked member cancels every lower live one with its children, slots and checks | | [0006](./0006-spend-guard.md) | proposed | A monthly budget closes admission and a burn ceiling caps the summed hourly rate of admitted runs; both default low in code | -| [0007](./0007-cost-efficient-execution.md) | proposed | Dispatch only opted-in work, upload each content-addressed cache key once, build heavy native dependencies once in the image | +| [0007](./0007-cost-efficient-execution.md) | proposed | Dispatch only opted-in work, upload each content-addressed cache key once, build heavy native dependencies once in the image, share a laned compile cache | | [0008](./0008-cheaper-waiting-and-transport.md) | proposed | rpc sandbox transport by default, admission polls scaled by queue position, one log line per supersession | | [0009](./0009-fast-trustworthy-verdicts.md) | proposed | Interactive admission class ahead of batch, budgeted and reconciled check-run delivery, a light pull-request tier with the heavy gate on the default branch, pool caps as upper bounds under the spend guard | diff --git a/wrangler.jsonc b/wrangler.jsonc index 512b55a..6945812 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -48,9 +48,9 @@ // every finished execution GitHub never received (DELIVERY_SWEEP_CRON, // apps/dispatcher/src/delivery-sweep.ts). "*/10 * * * *", - // The compile-cache sweep, not a run: measures each repository's - // compile-cache prefix, deletes expired entries and evicts down to the - // ceiling (COMPILE_CACHE_SWEEP_CRON, apps/dispatcher/src/routes/compile-cache.ts). + // The compile-cache sweep, not a run: deletes expired entries, evicts + // each repository down to its ceiling and measures it into the ledger + // (COMPILE_CACHE_SWEEP_CRON, apps/dispatcher/src/routes/compile-cache.ts). "41 * * * *", ], }, @@ -104,9 +104,11 @@ "ADMISSION_CAP_BROWSER": "16", "ADMISSION_CAP_RELEASE": "8", // Compile-cache bounds (apps/dispatcher/src/routes/compile-cache.ts): one - // entry, one repository prefix, and the entry age the hourly sweep deletes. + // entry, one repository, one execution's writes, and the entry age the + // hourly sweep deletes. "COMPILE_CACHE_MAX_OBJECT_BYTES": "104857600", "COMPILE_CACHE_MAX_REPO_BYTES": "21474836480", + "COMPILE_CACHE_MAX_EXECUTION_BYTES": "10737418240", "COMPILE_CACHE_TTL_DAYS": "14", // Opt the human-facing viewer surfaces (incl. the GitHub App install flow) // out of the Cloudflare Access gate for the initial dogfood. Configure @@ -248,6 +250,7 @@ { "name": "AGENT_BUDGET", "class_name": "AgentBudget" }, { "name": "RUNS_SANDBOX_AGENT", "class_name": "RunSandboxAgent" }, { "name": "RUNS_SANDBOX_RELEASE", "class_name": "RunSandboxRelease" }, + { "name": "COMPILE_CACHE_LEDGER", "class_name": "CompileCacheLedger" }, ], }, @@ -257,6 +260,7 @@ { "tag": "v2", "new_sqlite_classes": ["AgentBudget"] }, { "tag": "v3", "new_sqlite_classes": ["RunSandboxAgent"] }, { "tag": "v4", "new_sqlite_classes": ["RunSandboxRelease"] }, + { "tag": "v5", "new_sqlite_classes": ["CompileCacheLedger"] }, ], // Bucket name, database name and id, and namespace ids: operator overlay.