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.