diff --git a/apps/dispatcher/src/dashboard.ts b/apps/dispatcher/src/dashboard.ts index 0648ca79..ce53a7b2 100644 --- a/apps/dispatcher/src/dashboard.ts +++ b/apps/dispatcher/src/dashboard.ts @@ -52,6 +52,8 @@ export interface DashboardData { readonly nowMs: number; /** `owner/repo` slug for the source link. */ readonly repoSlug: string; + /** The spend guard's month-to-date line (`GET /api/spend`); absent when unreadable. */ + readonly spendLine?: string; } /** Server-side filters, parsed from the `run`/`repo`/`status` query params. */ @@ -196,6 +198,7 @@ const STYLE = ` header { display: flex; flex-wrap: wrap; align-items: baseline; gap: .75rem; margin-bottom: .25rem; } h1 { font-size: 1.4rem; margin: 0; letter-spacing: -0.01em; } .tagline { color: #9aa3b2; margin: 0 0 1.75rem; } + .spend { color: #9aa3b2; margin: -1.25rem 0 1.75rem; font-size: 0.9rem; } h2 { font-size: .82rem; text-transform: uppercase; letter-spacing: .08em; color: #9aa3b2; margin: 2rem 0 .6rem; } table { width: 100%; border-collapse: collapse; font-size: 14px; } th { text-align: left; font-weight: 600; color: #9aa3b2; padding: .4rem .5rem; border-bottom: 1px solid #232838; } @@ -375,7 +378,7 @@ export const renderDashboard = (data: DashboardData): string => {

FlareDispatch

Dispatch CI/CD runs to Cloudflare — inspect logs, executions, and product demos.

- +${data.spendLine !== undefined ? `

${esc(data.spendLine)}

\n` : ""}

Latest executions

${renderFilters(data.filters)} diff --git a/apps/dispatcher/src/env.ts b/apps/dispatcher/src/env.ts index a80de656..7baddfe2 100644 --- a/apps/dispatcher/src/env.ts +++ b/apps/dispatcher/src/env.ts @@ -464,6 +464,44 @@ export interface Env { */ readonly ADMISSION_CAP_AGENT?: string; + /** + * The browser pool's admission slots — a var, carried as a string. The + * browser pool admits at most the smaller of this and `ADMISSION_CAP`; keep + * it ≤ `RunSandboxBrowser`'s `max_instances`. Absent / unparsable → 16 + * (`BROWSER_ADMISSION_CAP_DEFAULT`). + */ + readonly ADMISSION_CAP_BROWSER?: string; + + /** + * The monthly spend budget in USD (specs/adr/0006-spend-guard.md) — a var. + * At or over it, admission skips every run with a neutral check. Unset or + * unparsable → 50; `unlimited` disables it. CONFIG_KV + * `spend.budgetMonthlyUsd` overrides it without a deploy. The default lives + * in code (core spend.ts), because an operator overlay replaces the + * committed wrangler.jsonc. + */ + readonly SPEND_BUDGET_MONTHLY_USD?: string; + + /** + * The burn ceiling in USD per hour — a var. The summed hourly rate of every + * admitted execution, across all pools, stays at or under it. Unset or + * unparsable → 1.50; `unlimited` disables it. CONFIG_KV + * `spend.burnMaxUsdPerHour` overrides it. + */ + readonly SPEND_BURN_MAX_USD_PER_HOUR?: string; + + /** + * Multiplier on metered container + model spend for Workflow steps and + * Durable Objects — a var. Unset or outside [1, 10] → 1.10. + */ + readonly SPEND_OVERHEAD_FACTOR?: string; + + /** + * Fraction of the monthly budget whose crossing sends the month's one + * warning on the notice path — a var. Unset or outside (0, 1] → 0.8. + */ + readonly SPEND_WARN_RATIO?: string; + /** * Cloudflare account id (the 32-hex id in the dashboard URL) — a var, not a * secret. When set, `RunWorkflow` builds a `details_url` on the GitHub diff --git a/apps/dispatcher/src/http-app.ts b/apps/dispatcher/src/http-app.ts index bdf32d75..f617b086 100644 --- a/apps/dispatcher/src/http-app.ts +++ b/apps/dispatcher/src/http-app.ts @@ -45,6 +45,7 @@ import { handleProductDemo } from "./routes/product-demos"; import { handleReplay } from "./routes/replay"; import { handleGithubWebhook } from "./routes/webhook"; import { handleSignalsWebhook } from "./routes/signals-webhook"; +import { handleSpend, readSpendNow, spendLine, spendView } from "./routes/spend"; /** `owner/repo` slug for the dashboard source link. */ const REPO_SLUG = "fractalbox/flare-dispatch"; @@ -368,6 +369,8 @@ const appShellRoute = Effect.gen(function* () { } const { origin, rows, filters, pagination } = yield* dashboardData; + // The spend line is a read beside the list, never a reason the page fails. + const spend = yield* readSpendNow(env).pipe(Effect.option); const html = renderDashboard({ origin, rows, @@ -375,6 +378,7 @@ const appShellRoute = Effect.gen(function* () { pagination, nowMs: Date.now(), repoSlug: REPO_SLUG, + ...(Option.isSome(spend) ? { spendLine: spendLine(spend.value) } : {}), }); return HttpServerResponse.raw(html, { status: 200, @@ -396,8 +400,13 @@ const dashboardJsonRoute = Effect.gen(function* () { if (request.method !== "GET") return methodNotAllowed; const { origin, rows, filters, pagination } = yield* dashboardData; + const env = yield* CurrentEnv; + const spend = yield* readSpendNow(env).pipe( + Effect.map((status) => ({ ...spendView(status), line: spendLine(status) })), + Effect.orElseSucceed(() => null), + ); return HttpServerResponse.raw( - JSON.stringify({ origin, repoSlug: REPO_SLUG, rows, filters, pagination }), + JSON.stringify({ origin, repoSlug: REPO_SLUG, rows, filters, pagination, spend }), { status: 200, headers: { "content-type": "application/json; charset=utf-8" }, @@ -585,6 +594,12 @@ const router = baseRouter.pipe( ), HttpRouter.all("/v1/dashboard.json", dashboardJsonRoute), HttpRouter.all("/v1/analytics.json", analyticsJsonRoute), + // The spend guard's month-to-date view (ADR-0006) — Access-gated like the + // executions and analytics reads. + HttpRouter.all( + "/api/spend", + route("GET", ({ env }) => handleSpend(env), { viewer: true }), + ), // Container-side OTP read — TOKEN-ONLY (no `viewer: true`): a sandbox can't // carry an Access JWT, so the expiring capability token + burn-after-read are // the gate. See routes/mailbox.ts for the posture. diff --git a/apps/dispatcher/src/routes/spend.test.ts b/apps/dispatcher/src/routes/spend.test.ts new file mode 100644 index 00000000..433bbcc1 --- /dev/null +++ b/apps/dispatcher/src/routes/spend.test.ts @@ -0,0 +1,129 @@ +// The spend guard's reader-facing surfaces: a budget or ceiling skip concludes +// the check neutral, and `GET /api/spend` / the dashboard line project one +// spend read (specs/adr/0006-spend-guard.md). + +import { Exit } from "effect"; +import { describe, expect, it } from "vitest"; +import { + decideBudget, + resolveSpendLimits, + RunSkipped, + spendCeilingReason, + spendMonth, +} from "@fractalboxdev/flare-dispatch-core"; +import type { SpendStatus } from "@fractalboxdev/flare-dispatch-runtime-cf/spend"; +import { runSkippedReason } from "../failure-summary"; +import { handleRequest } from "../router"; +import { makeFakeD1, makeFakeEnv, makeFakeR2, makeFakeWorkflow } from "../test-helpers"; +import { checkConclusion } from "../terminal-verdict"; +import { spendLine, spendView } from "./spend"; + +const NOW = Date.UTC(2026, 9, 22, 12, 0); + +const status = (monthToDateMicroUsd: number, admittedMicroUsdPerHour = 401_040): SpendStatus => { + const limits = resolveSpendLimits({}); + return { + month: spendMonth(NOW), + limits, + reading: { + finishedMicroUsd: monthToDateMicroUsd, + inFlightMicroUsd: 0, + admittedMicroUsdPerHour, + admittedCount: 1, + }, + monthToDateMicroUsd, + decision: decideBudget(monthToDateMicroUsd, limits), + }; +}; + +describe("a spend skip concludes neutral", () => { + it("records a budget skip as skipped with its reason, and a neutral check", () => { + const decision = decideBudget(50_500_000, resolveSpendLimits({})); + if (decision._kind !== "closed") throw new Error("expected a closed budget"); + const exit = Exit.fail(new RunSkipped({ reason: decision.reason })); + expect(runSkippedReason(exit)).toBe("spend budget reached: $50.50 of $50.00 this month"); + expect(checkConclusion("skipped", true)).toBe("neutral"); + }); + + it("records a burn-ceiling queue timeout as a skip", () => { + const reason = spendCeilingReason({ + admittedMicroUsdPerHour: 1_203_120, + ownMicroUsdPerHour: 401_040, + burnMaxMicroUsdPerHour: 1_500_000, + }); + expect(runSkippedReason(Exit.fail(new RunSkipped({ reason })))).toMatch(/^spend ceiling: /); + }); +}); + +describe("spendView", () => { + it("returns month-to-date spend, the limits, the admitted burn and the reset", () => { + expect(spendView(status(12_340_000))).toEqual({ + month: "2026-10", + monthToDateUsd: 12.34, + budgetUsd: 50, + burnMaxUsdPerHour: 1.5, + admittedUsdPerHour: 0.40104, + admittedCount: 1, + overheadFactor: 1.1, + warnRatio: 0.8, + budgetReached: false, + resetsAt: "2026-11-01T00:00:00.000Z", + }); + }); + + it("reports an unlimited limit as null", () => { + const open = status(1_000_000); + const view = spendView({ + ...open, + limits: { ...open.limits, budgetMicroUsd: null, burnMaxMicroUsdPerHour: null }, + }); + expect(view).toMatchObject({ budgetUsd: null, burnMaxUsdPerHour: null }); + }); +}); + +describe("GET /api/spend", () => { + const env = (viewerAccessMode?: "required") => + makeFakeEnv({ + hmacSecret: "s", + workflow: makeFakeWorkflow(), + storage: makeFakeR2(), + metadata: makeFakeD1({ executions: [] }), + ...(viewerAccessMode !== undefined ? { viewerAccessMode } : {}), + }); + + it("returns the deploy's spend view under the code defaults", async () => { + const res = await handleRequest(new Request("https://dispatcher.example.com/api/spend"), env()); + expect(res.status).toBe(200); + expect(await res.json()).toMatchObject({ + monthToDateUsd: 0, + budgetUsd: 50, + burnMaxUsdPerHour: 1.5, + budgetReached: false, + }); + }); + + it("is a GET viewer surface: Access gates it first, then the method", async () => { + const gated = await handleRequest( + new Request("https://dispatcher.example.com/api/spend"), + env("required"), + ); + expect(gated.status).toBe(503); + const posted = await handleRequest( + new Request("https://dispatcher.example.com/api/spend", { method: "POST" }), + env(), + ); + expect(posted.status).toBe(405); + }); +}); + +describe("spendLine", () => { + it("is the dashboard's one-line summary", () => { + expect(spendLine(status(12_340_000))).toBe( + "Spend 2026-10: $12.34 of $50.00 · burn $0.40/h of $1.50/h · resets 2026-11-01", + ); + }); + + it("says when the budget closed", () => { + expect(spendLine(status(55_000_000))).toContain("$55.00 of $50.00 — budget reached, runs skip"); + }); +}); diff --git a/apps/dispatcher/src/routes/spend.ts b/apps/dispatcher/src/routes/spend.ts new file mode 100644 index 00000000..c4f8cf19 --- /dev/null +++ b/apps/dispatcher/src/routes/spend.ts @@ -0,0 +1,92 @@ +// FlareDispatch Dispatcher — `GET /api/spend`: the spend guard's month-to-date view. +// +// Returns the deploy's month-to-date spend, the monthly budget, the burn +// ceiling, the admitted hourly rate and when the month resets — the same +// numbers admission decides with (runtime-cf spend-d1.ts). The dashboard +// renders `spendLine` of the same view. Access-gated like the executions and +// analytics reads. +// +// Spec: specs/adr/0006-spend-guard.md § Warning and visibility. + +import { Effect } from "effect"; +import { MICRO_USD_PER_USD, formatUsd } from "@fractalboxdev/flare-dispatch-core"; +import { readSpendLimits, readSpendStatus, type SpendStatus } from "@fractalboxdev/flare-dispatch-runtime-cf/spend"; +import type { Env } from "../env"; + +/** The JSON body of `GET /api/spend`. USD figures; `null` = unlimited. */ +export type SpendView = { + /** The UTC calendar month, `YYYY-MM`. */ + readonly month: string; + readonly monthToDateUsd: number; + readonly budgetUsd: number | null; + readonly burnMaxUsdPerHour: number | null; + /** The summed hourly rate of every admitted execution, across all pools. */ + readonly admittedUsdPerHour: number; + readonly admittedCount: number; + readonly overheadFactor: number; + readonly warnRatio: number; + /** `true` once month-to-date spend reaches the budget: runs skip. */ + readonly budgetReached: boolean; + /** The first instant of the next UTC month, ISO 8601. */ + readonly resetsAt: string; +}; + +const usd = (microUsd: number): number => microUsd / MICRO_USD_PER_USD; + +/** Project a spend read onto the API view. Pure. */ +export const spendView = (status: SpendStatus): SpendView => ({ + month: status.month.key, + monthToDateUsd: usd(status.monthToDateMicroUsd), + budgetUsd: status.limits.budgetMicroUsd === null ? null : usd(status.limits.budgetMicroUsd), + burnMaxUsdPerHour: + status.limits.burnMaxMicroUsdPerHour === null ? null : usd(status.limits.burnMaxMicroUsdPerHour), + admittedUsdPerHour: usd(status.reading.admittedMicroUsdPerHour), + admittedCount: status.reading.admittedCount, + overheadFactor: status.limits.overheadFactor, + warnRatio: status.limits.warnRatio, + budgetReached: status.decision._kind === "closed", + resetsAt: new Date(status.month.resetsAtMs).toISOString(), +}); + +/** The one-line summary the dashboard shows. Pure. */ +export const spendLine = (status: SpendStatus): string => { + const budget = status.limits.budgetMicroUsd; + const ceiling = status.limits.burnMaxMicroUsdPerHour; + return [ + `Spend ${status.month.key}: ${formatUsd(status.monthToDateMicroUsd)} of ${ + budget === null ? "unlimited" : formatUsd(budget) + }${status.decision._kind === "closed" ? " — budget reached, runs skip" : ""}`, + `burn ${formatUsd(status.reading.admittedMicroUsdPerHour)}/h of ${ + ceiling === null ? "unlimited" : `${formatUsd(ceiling)}/h` + }`, + `resets ${new Date(status.month.resetsAtMs).toISOString().slice(0, 10)}`, + ].join(" · "); +}; + +/** Read the deploy's spend status now. */ +export const readSpendNow = (env: Env, nowMs: number = Date.now()): Effect.Effect => + readSpendLimits(env.CONFIG_KV, env).pipe( + Effect.flatMap((limits) => readSpendStatus(env.RUNS_METADATA, limits, nowMs)), + ); + +/** `GET /api/spend`. A D1 read failure answers 503 with the reason. */ +export const handleSpend = (env: Env, nowMs: number = Date.now()): Promise => + Effect.runPromise( + readSpendNow(env, nowMs).pipe( + Effect.map( + (status) => + new Response(JSON.stringify(spendView(status)), { + status: 200, + headers: { "content-type": "application/json; charset=utf-8" }, + }), + ), + Effect.catchAll((error) => + Effect.succeed( + new Response(JSON.stringify({ error: "spend_unreadable", detail: error.message }), { + status: 503, + headers: { "content-type": "application/json" }, + }), + ), + ), + ), + ); diff --git a/apps/dispatcher/src/sandbox-routing.test.ts b/apps/dispatcher/src/sandbox-routing.test.ts index 4b49a16e..1ecf0fda 100644 --- a/apps/dispatcher/src/sandbox-routing.test.ts +++ b/apps/dispatcher/src/sandbox-routing.test.ts @@ -55,15 +55,39 @@ const POOL_CLASS = { lean: "RunSandbox", browser: "RunSandboxBrowser", agent: "R const bound = { RUNS_SANDBOX_BROWSER: browser, RUNS_SANDBOX_AGENT: agent, RUNS_SANDBOX_RELEASE: release }; describe("selectAdmission — the pool a run is metered against and its slot cap", () => { - it("admits no pool past its container's max_instances under the committed vars", () => { + it("keeps every pool's cap below its container's max_instances under the committed vars", () => { for (const image of ["lean", "browser", "agent", "release"] as const) { const { pool, cap } = selectAdmission(image, { ...wrangler.vars, ...bound }); expect(pool).toBe(image); const container = wrangler.containers.find((entry) => entry.class_name === POOL_CLASS[pool])!; - expect(cap, `${pool} pool`).toBeLessThanOrEqual(container.max_instances); + expect(cap, `${pool} pool`).toBeLessThan(container.max_instances); } }); + it("keeps every pool's cap below its container's max_instances on the code defaults", () => { + for (const image of ["lean", "browser", "agent", "release"] as const) { + const { pool, cap } = selectAdmission(image, bound); + const container = wrangler.containers.find((entry) => entry.class_name === POOL_CLASS[pool])!; + expect(cap, `${pool} pool`).toBeLessThan(container.max_instances); + } + }); + + it("meters every sandbox container class as one pool", () => { + const sandboxClasses = wrangler.containers + .map((entry) => entry.class_name) + .filter((name) => name.startsWith("RunSandbox")) + .sort(); + expect(sandboxClasses).toEqual(Object.values(POOL_CLASS).sort()); + }); + + it("caps the browser pool by ADMISSION_CAP_BROWSER", () => { + expect(wrangler.vars.ADMISSION_CAP_BROWSER).toBeDefined(); + expect(selectAdmission("browser", { ADMISSION_CAP: "16", ADMISSION_CAP_BROWSER: "4", ...bound })).toEqual({ + pool: "browser", + cap: 4, + }); + }); + it("caps the release and agent pools at their eight containers", () => { expect(selectAdmission("release", { ...wrangler.vars, ...bound })).toEqual({ pool: "release", cap: 8 }); expect(selectAdmission("agent", { ...wrangler.vars, ...bound })).toEqual({ pool: "agent", cap: 8 }); diff --git a/apps/dispatcher/src/workflow.ts b/apps/dispatcher/src/workflow.ts index f4fd7c81..32059378 100644 --- a/apps/dispatcher/src/workflow.ts +++ b/apps/dispatcher/src/workflow.ts @@ -61,6 +61,9 @@ import { getSandbox } from "@cloudflare/sandbox"; import { Duration, Effect, Exit, identity, Runtime, Schedule, Schema } from "effect"; import { admissionAcquireAttempts, + ADMISSION_CLASS_DEFAULT, + admissionJitterMs, + admissionPollSleepMs, AdmissionTimedOut, Checks, decideAdmission, @@ -72,7 +75,10 @@ import { isEngineInterruptedCause, isEngineInterruption, withRunCeiling, - type RunSkipped, + RunSkipped, + spendCeilingReason, + type AdmissionObservation, + type SpendLimits, type RunContext, type RunError, } from "@fractalboxdev/flare-dispatch-core"; @@ -84,6 +90,7 @@ import { type ChecksGithubConfig, describeOutcome, type EmailCloudflareConfig, + hourlyRateForSandboxImage, instanceForSandboxImage, LEASE_HEARTBEAT_EVERY_MS, makeCFRuntimeLive, @@ -94,9 +101,13 @@ import { makeSerialQueueD1, makeWritebackTokenMinter, previewSafeSandboxId, + readSpendLimits, + readSpendStatus, recordExecutionCost, runSerialGate, runWriteback, + type SpendStatus, + warnSpendOnce, type WritebackOutcome, } from "@fractalboxdev/flare-dispatch-runtime-cf"; import { WRITEBACK_ARTIFACT } from "@fractalboxdev/flare-dispatch-core"; @@ -434,9 +445,10 @@ export class RunWorkflow extends WorkflowEntrypoint { // selection above (browser/agent/release image without a bound container // degrades to lean), so a run is always metered against the pool whose // container it will actually consume — and that pool's slot cap: the plain - // `ADMISSION_CAP` wrangler var (unset/unparsable → 16), and for the agent - // and release pools the smaller of that and `ADMISSION_CAP_AGENT` / - // `ADMISSION_CAP_RELEASE` (unset → 8). Keep every cap ≤ its container's + // `ADMISSION_CAP` wrangler var (unset/unparsable → 16), and for the agent, + // browser and release pools the smaller of that and `ADMISSION_CAP_AGENT` / + // `ADMISSION_CAP_BROWSER` / `ADMISSION_CAP_RELEASE` (unset → 8, 16, 8). + // Keep every cap ≤ its container's // `max_instances` (sandbox-routing.test.ts reads wrangler.jsonc to check). const { pool: admissionPool, cap: admissionCap } = selectAdmission(run.sandboxImage, this.env); @@ -614,6 +626,17 @@ export class RunWorkflow extends WorkflowEntrypoint { // The `Workflow` binding backs the `childRuns` capability — a run can // `spawnChildRun` / `fanOut` to independent child `RunWorkflow` instances. runsWorkflow: this.env.RUNS_WORKFLOW, + // A fan-out parent reads spend before it spawns children: over the + // monthly budget it skips and spawns none (ADR-0006). `fanOut` calls + // this inside its spawn step, so a replay reads the checkpoint. + spendGate: async () => { + const status = await Effect.runPromise( + readSpendLimits(this.env.CONFIG_KV, this.env).pipe( + Effect.flatMap((limits) => readSpendStatus(this.env.RUNS_METADATA, limits, Date.now())), + ), + ); + return status.decision._kind === "closed" ? status.decision.reason : undefined; + }, executionId: payload.executionId, execution: { repo: payload.github.repo, @@ -878,17 +901,74 @@ export class RunWorkflow extends WorkflowEntrypoint { }); return yield* Effect.fail(error); }); + // --- Spend guard (specs/adr/0006-spend-guard.md) ----------------------- + // + // Admission reads month-to-date spend before it enqueues and again + // before each claim; at or over the monthly budget the run ends + // `RunSkipped` — a neutral check, a `skipped` row. The claim itself + // enforces the burn ceiling across every pool, at this run's container + // class rate. Every read sits in a `step.do`, so a replay reads the + // checkpointed answer. + const hourlyRateMicroUsd = hourlyRateForSandboxImage(admissionPool); + const spendCheck = (name: string, limits: SpendLimits) => + Effect.gen(function* () { + const status: SpendStatus = yield* stepDo(name, () => + Effect.runPromise(readSpendStatus(db, limits, Date.now())), + ); + if (status.decision._kind === "closed") { + yield* Effect.logWarning(`spend.closed ${payload.executionId} — ${status.decision.reason}`); + return yield* Effect.fail(new RunSkipped({ reason: status.decision.reason })); + } + return status; + }); const admissionGate: Effect.Effect = Effect.gen(function* () { + const limits = yield* stepDo("spend-limits", () => + Effect.runPromise(readSpendLimits(env.CONFIG_KV, env)), + ); + let spend = yield* spendCheck("spend-check-enqueue", limits); + const burnMaxMicroUsdPerHour = limits.burnMaxMicroUsdPerHour; + // The run's admission class (ADR-0009): interactive waiters are + // admitted ahead of batch ones in the same pool. The burn ceiling + // applies to both and never reorders the line. + const claimer = makeRunAdmissionD1( + db, + Date.now, + admissionCap, + { hourlyRateMicroUsd, burnMaxMicroUsdPerHour }, + run.limits.admissionClass ?? ADMISSION_CLASS_DEFAULT, + ); + // A queue timeout while the burn ceiling, not the pool cap, held the + // run back is a spend skip; under the cap alone it stays a failure. + const queueTimeout = ( + observed: AdmissionObservation & { readonly now: number }, + timedOut: AdmissionTimedOut, + ): Effect.Effect => + observed.ceilingHeld === true && burnMaxMicroUsdPerHour !== null + ? Effect.fail( + new RunSkipped({ + reason: spendCeilingReason({ + admittedMicroUsdPerHour: spend.reading.admittedMicroUsdPerHour, + ownMicroUsdPerHour: hourlyRateMicroUsd, + burnMaxMicroUsdPerHour, + }), + }), + ) + : admissionTimedOut(timedOut, enqueuedAt, observed.now); + // Enqueue — the row's `enqueued_at` (FIFO key + dispatch-age basis) // is read inside the step, so a replay sees the SAME timestamp. const { enqueuedAt } = yield* stepDo("admission-enqueue", () => Effect.runPromise( - admissions.enqueue(payload.executionId, admissionPool, payload.parentExecutionId), + claimer.enqueue(payload.executionId, admissionPool, payload.parentExecutionId), ), ); // Bounded claim/sleep loop uses a count, so the bound is replay-stable - // and trivial against the Workflows step cap (#83 precedent). + // and trivial against the Workflows step cap (#83 precedent). Each + // attempt is three steps (spend check, claim, sleep). The count is + // derived from the SHORTEST sleep, so it covers the queue age however + // the sleeps scale; a waiter sleeping longer reaches the age limit, + // read inside its claim step, in fewer attempts. const maxAttempts = admissionAcquireAttempts( admissionMaxQueueAgeMs, ADMISSION_POLL_EVERY_MS, @@ -899,12 +979,13 @@ export class RunWorkflow extends WorkflowEntrypoint { // A child reads its parent's row before each claim: once a parent // that joins its children has ended, the child leaves the queue // instead (admission-parent.ts). + spend = yield* spendCheck(`spend-check-${i}`, limits); const observed = yield* stepDo(`admission-claim-${i}`, () => claimUnlessParentEnded({ parentExecutionId: payload.parentExecutionId, parentRow: (id) => Effect.runPromise(supersessions.row(id)), joinsChildren: (name) => lookupRun(name)?.joinsChildren === true, - claim: () => Effect.runPromise(admissions.attempt(payload.executionId, admissionPool, enqueuedAt)), + claim: () => Effect.runPromise(claimer.attempt(payload.executionId, admissionPool, enqueuedAt)), }), ); if (isParentEnded(observed)) { @@ -941,16 +1022,42 @@ export class RunWorkflow extends WorkflowEntrypoint { admissionMaxQueueAgeMs, ); - if (decision._kind === "admit") return; + if (decision._kind === "admit") { + // The month's one warning, past `SPEND_WARN_RATIO` of the budget. + const status = spend; + yield* stepDo("spend-warn", () => + Effect.runPromise( + warnSpendOnce({ + kv: env.CONFIG_KV, + status, + notify: async (text, monthKey) => { + const sent = await emitSlackNotice(env, { + useCase: "spend-warn", + deliveryId: `spend-warn-${monthKey}`, + text, + run: payload.run, + executionId: payload.executionId, + }); + if (sent.outcome === "failed") throw new Error(sent.reason); + }, + }), + ), + ); + // Spread a burst of admissions over up to two seconds before the + // container starts (ADR-0009). Derived from the execution id and + // slept durably, so a replay neither re-draws nor re-sleeps it. + const jitterMs = admissionJitterMs(payload.executionId); + if (jitterMs > 0) yield* stepSleep("admission-jitter", jitterMs); + return; + } if (decision._kind === "timeout") { - return yield* admissionTimedOut( + return yield* queueTimeout( + observed, new AdmissionTimedOut({ queuedForMs: decision.queuedForMs, position: decision.position, poolBusy: decision.poolBusy, }), - enqueuedAt, - observed.now, ); } @@ -982,19 +1089,29 @@ export class RunWorkflow extends WorkflowEntrypoint { ); } - // Durable, free hibernation between polls. Don't sleep after the - // final attempt — fall straight through to timeout. + // Durable, free hibernation between polls, scaled with the position + // the claim step observed (ADR-0008): 20 s at the head, up to 60 s + // several slots back, never past the age limit by more than one + // 20 s poll. Both inputs are checkpointed, so a replay sleeps the + // same length. Don't sleep after the final attempt — fall straight + // through to timeout. if (i < maxAttempts - 1) { - yield* stepSleep(`admission-wait-${i}`, ADMISSION_POLL_EVERY_MS); + yield* stepSleep( + `admission-wait-${i}`, + admissionPollSleepMs( + decision.position, + admissionCap, + enqueuedAt + admissionMaxQueueAgeMs - observed.now, + ), + ); } else { - return yield* admissionTimedOut( + return yield* queueTimeout( + observed, new AdmissionTimedOut({ queuedForMs: Math.max(0, observed.now - enqueuedAt), position: decision.position, poolBusy: decision.poolBusy, }), - enqueuedAt, - observed.now, ); } } @@ -1343,13 +1460,14 @@ export class RunWorkflow extends WorkflowEntrypoint { // Per-execution cost rollup — METERED model tokens (summed from // `execution_model_usage`, written by the modelGateway wrapper) + MODELED - // container compute (instance × wall-time). Runs AFTER finishExecution so - // the wall-time it reads is complete. Best-effort: a cost failure logs and - // never flips the verdict (recordExecutionCost absorbs its own errors). + // container compute (the admission pool's instance list rate × admitted + // seconds, ADR-0006 § Meter). Runs AFTER finishExecution so the window + // it reads is closed. Best-effort: a cost failure logs and never flips + // the verdict (recordExecutionCost absorbs its own errors). yield* recordExecutionCost({ db, executionId: payload.executionId, - instance: instanceForSandboxImage(run.sandboxImage), + instance: instanceForSandboxImage(admissionPool), }); // --- Post-run writeback (the Worker proposes the diff as a PR) -------- diff --git a/apps/docs/src/data/benchmarks.ts b/apps/docs/src/data/benchmarks.ts index 414c9ca9..83b50484 100644 --- a/apps/docs/src/data/benchmarks.ts +++ b/apps/docs/src/data/benchmarks.ts @@ -11,10 +11,9 @@ // and diffs. Model-calling runs are priced on the Anthropic backend as a representative // metered figure; the default Workers AI catalog backend bills as account-level Neurons. import { - CONTAINER_GIB_MICRO_USD_PER_SEC, - CONTAINER_VCPU_MICRO_USD_PER_SEC, INSTANCE_SPECS, containerCostMicroUsd, + containerMicroUsdPerSec, modelCost, type InstanceType, } from "@fractalboxdev/flare-dispatch-core/cost"; @@ -200,9 +199,4 @@ export const benchmarks: readonly BenchmarkRow[] = PROFILES.map((p) => { }); /** Container rate for an instance, in micro-USD per active second. */ -export const containerMicroUsdPerSec = (instance: InstanceType): number => { - const spec = INSTANCE_SPECS[instance]; - return spec.vcpu * CONTAINER_VCPU_MICRO_USD_PER_SEC + spec.gib * CONTAINER_GIB_MICRO_USD_PER_SEC; -}; - -export { INSTANCE_SPECS }; +export { containerMicroUsdPerSec, INSTANCE_SPECS }; diff --git a/infra/migrations/0011_spend_guard.sql b/infra/migrations/0011_spend_guard.sql new file mode 100644 index 00000000..d8671064 --- /dev/null +++ b/infra/migrations/0011_spend_guard.sql @@ -0,0 +1,34 @@ +-- Migration number: 0011 spend guard +-- +-- The meter, the monthly budget and the burn ceiling of +-- specs/adr/0006-spend-guard.md, and the interactive admission class of +-- specs/adr/0009-fast-trustworthy-verdicts.md: +-- +-- * `executions.admitted_at` — when the run's admission claim landed. A +-- container is active from admission to completion, so the finish path +-- prices that window (runtime-cf execution-cost.ts) and queue wait costs +-- nothing. NULL for a run never admitted. +-- * `executions.cost_meter` — which rule priced `cost_micro_usd`: +-- `admitted-active` from here on; rows priced on wall time before this +-- migration keep their value and read `modeled-wall`. +-- * `run_admissions.hourly_rate_micro_usd` — the list rate of the row's +-- container class per hour. The claim admits only while the live admitted +-- rates plus the claimant's own fit under the burn ceiling, across every +-- pool (runtime-cf run-admission-d1.ts). +-- * `run_admissions.admission_class` — the waiter's line +-- (specs/adr/0009-fast-trustworthy-verdicts.md): the claim admits the +-- oldest live `interactive` waiter in a pool before any `batch` waiter, +-- and FIFO order and the queue position apply within a class. A NULL row +-- (pre-0011) ranks as `batch`. +-- +-- The month-to-date spend read (runtime-cf spend-d1.ts) sums `cost_micro_usd` +-- over executions admitted this month; `executions_admitted_at` serves that +-- range scan. + +ALTER TABLE executions ADD COLUMN admitted_at INTEGER; -- ms epoch; NULL = never admitted +ALTER TABLE executions ADD COLUMN cost_meter TEXT; -- admitted-active | modeled-wall +UPDATE executions SET cost_meter = 'modeled-wall' WHERE cost_micro_usd IS NOT NULL; +CREATE INDEX IF NOT EXISTS executions_admitted_at ON executions(admitted_at); + +ALTER TABLE run_admissions ADD COLUMN hourly_rate_micro_usd INTEGER; -- NULL rows (pre-0011) count as 0 +ALTER TABLE run_admissions ADD COLUMN admission_class TEXT; -- interactive | batch; NULL = batch diff --git a/packages/core/src/cost.test.ts b/packages/core/src/cost.test.ts index 1d079f8d..9b21d5db 100644 --- a/packages/core/src/cost.test.ts +++ b/packages/core/src/cost.test.ts @@ -5,11 +5,13 @@ import { describe, expect, it } from "vitest"; import { + CONTAINER_DISK_MICRO_USD_PER_SEC, CONTAINER_GIB_MICRO_USD_PER_SEC, CONTAINER_VCPU_MICRO_USD_PER_SEC, INSTANCE_SPECS, MICRO_USD_PER_USD, containerCostMicroUsd, + containerMicroUsdPerHour, estimateExecutionCost, formatMicroUsd, microUsdToUsd, @@ -18,13 +20,25 @@ import { } from "./cost.js"; describe("rate card (pinned to specs/06-cost.md)", () => { - it("standard-2 is 1 vCPU / 6 GiB", () => { - expect(INSTANCE_SPECS["standard-2"]).toEqual({ vcpu: 1, gib: 6 }); + it("standard-2 is 1 vCPU / 6 GiB / 12 GB disk", () => { + expect(INSTANCE_SPECS["standard-2"]).toEqual({ vcpu: 1, gib: 6, diskGb: 12 }); }); - it("basic is 1/4 vCPU / 1 GiB and standard-3 is 2 vCPU / 8 GiB", () => { - expect(INSTANCE_SPECS.basic).toEqual({ vcpu: 1 / 4, gib: 1 }); - expect(INSTANCE_SPECS["standard-3"]).toEqual({ vcpu: 2, gib: 8 }); + it("basic, standard-3 and standard-4 carry their provisioned disk", () => { + expect(INSTANCE_SPECS.basic).toEqual({ vcpu: 1 / 4, gib: 1, diskGb: 4 }); + expect(INSTANCE_SPECS["standard-3"]).toEqual({ vcpu: 2, gib: 8, diskGb: 16 }); + expect(INSTANCE_SPECS["standard-4"]).toEqual({ vcpu: 4, gib: 12, diskGb: 20 }); + }); + + it("disk is $0.00000007/GB-s in micro-USD", () => { + expect(CONTAINER_DISK_MICRO_USD_PER_SEC).toBe(0.07); + }); + + it("prices an hour on the provisioned size", () => { + // standard-4: 4 × 0.072 + 12 × 0.009 + 20 × 0.000252 = $0.40104/h. + expect(containerMicroUsdPerHour("standard-4")).toBe(401_040); + // standard-2: 0.072 + 6 × 0.009 + 12 × 0.000252 = $0.129024/h. + expect(containerMicroUsdPerHour("standard-2")).toBe(129_024); }); it("container rates are $0.000020/vCPU-s and $0.0000025/GiB-s in micro-USD", () => { @@ -35,17 +49,18 @@ describe("rate card (pinned to specs/06-cost.md)", () => { }); describe("containerCostMicroUsd — spec-06 per-execution anatomy", () => { - it("standard-2 for an 8-minute run ≈ $0.0168 (the spec's ~$0.017 example)", () => { + it("standard-2 for an 8-minute run ≈ $0.0172 (the spec's ~$0.017 example)", () => { // specs/06-cost.md § Per-execution cost anatomy: a standard-2 (1 vCPU, - // 6 GiB) running 8 min (480 s) = ~$0.0096 vCPU + ~$0.0072 mem ≈ $0.0168. + // 6 GiB, 12 GB) running 8 min (480 s) = $0.0096 vCPU + $0.0072 mem + + // $0.000403 disk ≈ $0.0172. const micro = containerCostMicroUsd({ instance: "standard-2", activeSeconds: 480 }); - expect(micro).toBe(16_800); - expect(microUsdToUsd(micro)).toBeCloseTo(0.0168, 6); + expect(micro).toBe(17_203); + expect(microUsdToUsd(micro)).toBeCloseTo(0.0172, 4); }); it("scales linearly with active seconds", () => { - const a = containerCostMicroUsd({ instance: "standard-2", activeSeconds: 60 }); - const b = containerCostMicroUsd({ instance: "standard-2", activeSeconds: 120 }); + const a = containerCostMicroUsd({ instance: "standard-2", activeSeconds: 100 }); + const b = containerCostMicroUsd({ instance: "standard-2", activeSeconds: 200 }); expect(b).toBe(a * 2); }); @@ -144,7 +159,7 @@ describe("estimateExecutionCost — honesty basis", () => { }); expect(cost.basis).toBe("modeled"); expect(cost.modelMicroUsd).toBeNull(); - expect(cost.totalMicroUsd).toBe(16_800); + expect(cost.totalMicroUsd).toBe(17_203); }); it("metered model with no container → metered", () => { diff --git a/packages/core/src/cost.ts b/packages/core/src/cost.ts index ba99be5b..580119d5 100644 --- a/packages/core/src/cost.ts +++ b/packages/core/src/cost.ts @@ -39,10 +39,11 @@ export const MICRO_USD_PER_USD = 1_000_000; // --- Container instance types --------------------------------------------------- // -// vCPU + memory per Cloudflare Containers instance type (specs/05-byoc.md § -// Wrangler config, "Instance types (2026-05)"). A run's instance type is fixed -// by the Durable Object class its `sandboxImage` selects (wrangler.jsonc): -// `lean`/`browser` → `standard-2`, `agent` → `standard-3`. +// Provisioned vCPU, memory and disk per Cloudflare Containers instance type +// (specs/05-byoc.md § Wrangler config, "Instance types (2026-05)"). A run's +// instance type is fixed by the container class its `sandboxImage` selects in +// wrangler.jsonc — the one table is `SANDBOX_CONTAINER` in runtime-cf +// execution-cost.ts, checked against that file by its test. export type InstanceType = | "lite" @@ -57,30 +58,54 @@ export type InstanceSpec = { readonly vcpu: number; /** Allotted memory in GiB. */ readonly gib: number; + /** Provisioned disk in GB. */ + readonly diskGb: number; }; -/** vCPU / memory by instance type (specs/05-byoc.md § Wrangler config). */ +/** vCPU / memory / disk by instance type (specs/05-byoc.md § Wrangler config). */ export const INSTANCE_SPECS: Record = { - lite: { vcpu: 1 / 16, gib: 0.25 }, - basic: { vcpu: 1 / 4, gib: 1 }, - "standard-1": { vcpu: 1 / 2, gib: 4 }, - "standard-2": { vcpu: 1, gib: 6 }, - "standard-3": { vcpu: 2, gib: 8 }, - "standard-4": { vcpu: 4, gib: 12 }, + lite: { vcpu: 1 / 16, gib: 0.25, diskGb: 2 }, + basic: { vcpu: 1 / 4, gib: 1, diskGb: 4 }, + "standard-1": { vcpu: 1 / 2, gib: 4, diskGb: 8 }, + "standard-2": { vcpu: 1, gib: 6, diskGb: 12 }, + "standard-3": { vcpu: 2, gib: 8, diskGb: 16 }, + "standard-4": { vcpu: 4, gib: 12, diskGb: 20 }, } as const; // --- Container rates (specs/06-cost.md § Pricing model) ------------------------- // -// $0.000020 / vCPU-s and $0.0000025 / GiB-s. In µ$: multiply USD by 1e6. +// List prices on the PROVISIONED size: $0.000020 / vCPU-s, $0.0000025 / GiB-s +// and $0.00000007 / GB-s of disk. In µ$: multiply USD by 1e6. // $0.000020 × 1e6 = 20 µ$ per vCPU-second // $0.0000025 × 1e6 = 2.5 µ$ per GiB-second -// (Disk GB-s is billed too but rounds to zero per-execution and we don't track -// it — see specs/06-cost.md § Per-execution cost anatomy.) +// $0.00000007 × 1e6 = 0.07 µ$ per disk GB-second +// Cloudflare bills active vCPU, so a provisioned-size rate is an upper bound on +// the billed rate — the direction a spend cap wants (ADR-0006 § Meter). /** Container vCPU-second rate, in micro-USD (`$0.000020/vCPU-s`). */ export const CONTAINER_VCPU_MICRO_USD_PER_SEC = 20; /** Container memory GiB-second rate, in micro-USD (`$0.0000025/GiB-s`). */ export const CONTAINER_GIB_MICRO_USD_PER_SEC = 2.5; +/** Container disk GB-second rate, in micro-USD (`$0.00000007/GB-s`). */ +export const CONTAINER_DISK_MICRO_USD_PER_SEC = 0.07; + +/** One instance's list rate per active second, in (fractional) micro-USD. */ +export const containerMicroUsdPerSec = (instance: InstanceType): number => { + const spec = INSTANCE_SPECS[instance]; + return ( + spec.vcpu * CONTAINER_VCPU_MICRO_USD_PER_SEC + + spec.gib * CONTAINER_GIB_MICRO_USD_PER_SEC + + spec.diskGb * CONTAINER_DISK_MICRO_USD_PER_SEC + ); +}; + +/** + * One instance's list rate per hour, in whole micro-USD — the unit the burn + * ceiling sums across admitted executions (`run_admissions.hourly_rate_micro_usd`). + * `standard-4` ≈ $0.401/h, `standard-2` ≈ $0.129/h. + */ +export const containerMicroUsdPerHour = (instance: InstanceType): number => + Math.round(containerMicroUsdPerSec(instance) * 3600); // --- Model token rates ---------------------------------------------------------- // @@ -209,23 +234,16 @@ export const modelCost = (opts: { }; /** - * Modeled container-compute cost: `(vCPU-s × vCPU-rate) + (GiB-s × GiB-rate)`. - * GiB-seconds derive from the instance's allotted memory × active seconds. This - * is ALWAYS modeled — Cloudflare meters container vCPU-s at the account level - * only, never per-execution. + * Modeled container-compute cost: `(vCPU-s × vCPU-rate) + (GiB-s × GiB-rate) + + * (disk GB-s × disk rate)`, each on the instance's provisioned size × active + * seconds. This is ALWAYS modeled — Cloudflare meters container usage at the + * account level only, never per-execution. */ export const containerCostMicroUsd = (opts: { readonly instance: InstanceType; - /** Seconds the container was active (boot → destroy, or summed exec time). */ + /** Seconds the container was active (admission → completion). */ readonly activeSeconds: number; -}): number => { - const spec = INSTANCE_SPECS[opts.instance]; - const vcpuSeconds = spec.vcpu * opts.activeSeconds; - const gibSeconds = spec.gib * opts.activeSeconds; - return Math.round( - vcpuSeconds * CONTAINER_VCPU_MICRO_USD_PER_SEC + gibSeconds * CONTAINER_GIB_MICRO_USD_PER_SEC, - ); -}; +}): number => Math.round(containerMicroUsdPerSec(opts.instance) * opts.activeSeconds); // --- Whole-execution estimate --------------------------------------------------- diff --git a/packages/core/src/define-run.ts b/packages/core/src/define-run.ts index d422d4a9..3a0cd427 100644 --- a/packages/core/src/define-run.ts +++ b/packages/core/src/define-run.ts @@ -9,12 +9,19 @@ import type { Effect, Schema } from "effect"; import type { RunContext } from "./context"; import type { RunError } from "./errors"; +import type { AdmissionClass } from "./run-admission"; import type { WritebackSpec } from "./writeback"; export type RunLimits = { readonly maxDurationSec: number; /** Maximum admission wait; defaults to the dispatcher's standard queue age. */ readonly admissionMaxQueueAgeSec?: number | ((input: I) => number); + /** + * The admission class this run waits in; `interactive` runs are admitted + * ahead of `batch` ones in the same pool. Defaults to `batch` + * (specs/adr/0009-fast-trustworthy-verdicts.md). + */ + readonly admissionClass?: AdmissionClass; readonly maxConcurrency?: number; readonly requiresBrowser?: boolean; }; diff --git a/packages/core/src/fakes/child-runs-fake.ts b/packages/core/src/fakes/child-runs-fake.ts index 45117653..6deb0ba3 100644 --- a/packages/core/src/fakes/child-runs-fake.ts +++ b/packages/core/src/fakes/child-runs-fake.ts @@ -8,7 +8,7 @@ // Spec: specs/03-dsl.md § spawnChildRun, specs/pm/plan.md § 3 (fakes/). import { Effect, Layer } from "effect"; -import { ChildSpawnFailed } from "../errors"; +import { ChildSpawnFailed, RunSkipped } from "../errors"; import { type ChildRunStatus, ChildRuns, @@ -49,6 +49,12 @@ export const makeChildRunsFake = (opts?: { existing?: Iterable; /** Reject selected spawns before recording them, for partial fan-out tests. */ failSpawn?: (opts: SpawnChildRunOpts) => boolean; + /** + * When set, `spendGate` fails `RunSkipped` with this reason (budget reached). + * A function is read on every gate call, so a test closes the budget mid-run; + * it returns `undefined` while the budget is open. + */ + spendClosed?: string | (() => string | undefined); /** * Static per-id poll statuses. An id absent from the map polls as `missing`. * For a loop test that flips `running → success`, supply `pollFn` instead. @@ -77,6 +83,11 @@ export const makeChildRunsFake = (opts?: { }; const service: ChildRunsService = { + spendGate: () => + Effect.suspend(() => { + const closed = typeof opts?.spendClosed === "function" ? opts.spendClosed() : opts?.spendClosed; + return closed !== undefined ? Effect.fail(new RunSkipped({ reason: closed })) : Effect.void; + }), handoffAdmission: () => Effect.sync(() => { state.admissionHandoffs += 1; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 4512ebec..44ef60e6 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -82,11 +82,44 @@ export { export { decideAdmission, admissionAcquireAttempts, + admissionPollSleepMs, + admissionJitterMs, + ADMISSION_CLASS_DEFAULT, + ADMISSION_POLL_MIN_MS, + ADMISSION_POLL_MAX_MS, + ADMISSION_JITTER_MAX_MS, + type AdmissionClass, type AdmissionPool, type AdmissionObservation, type AdmissionDecision, } from "./run-admission"; +// --- Spend guard (pure budget + burn-ceiling math, ADR-0006) ----------------- +export { + SPEND_BUDGET_MONTHLY_USD_DEFAULT, + SPEND_BURN_MAX_USD_PER_HOUR_DEFAULT, + SPEND_OVERHEAD_FACTOR_DEFAULT, + SPEND_WARN_RATIO_DEFAULT, + SPEND_UNLIMITED, + SPEND_BUDGET_KV_KEY, + SPEND_BURN_KV_KEY, + spendWarnedKvKey, + parseUsdLimit, + resolveSpendLimits, + spendMonth, + monthToDateMicroUsd, + formatUsd, + decideBudget, + fitsBurnCeiling, + spendCeilingReason, + type SpendLimits, + type SpendVars, + type SpendOverrides, + type SpendMonth, + type SpendReading, + type BudgetDecision, +} from "./spend"; + // --- Run ceiling (limits.maxDurationSec, replay-stable) ---------------------- export { withRunCeiling, type RunCeiling } from "./run-ceiling"; export { @@ -303,6 +336,9 @@ export { INSTANCE_SPECS, CONTAINER_VCPU_MICRO_USD_PER_SEC, CONTAINER_GIB_MICRO_USD_PER_SEC, + CONTAINER_DISK_MICRO_USD_PER_SEC, + containerMicroUsdPerSec, + containerMicroUsdPerHour, modelRate, modelCost, containerCostMicroUsd, diff --git a/packages/core/src/primitives/fan-out.test.ts b/packages/core/src/primitives/fan-out.test.ts index f7365961..fb1ee87e 100644 --- a/packages/core/src/primitives/fan-out.test.ts +++ b/packages/core/src/primitives/fan-out.test.ts @@ -82,3 +82,24 @@ describe("fanOut", () => { expect(spawned).toEqual([]); }); }); + +describe("fanOut — the spend gate (ADR-0006)", () => { + const reason = "spend budget reached: $90.00 of $90.00 this month"; + + it("skips and spawns no child once the monthly budget is reached", async () => { + const child = makeChildRunsFake({ spendClosed: reason }); + const error = await Effect.runPromise( + fanOut({ run: "check", items: ["a", "b"], toInput: (item) => ({ item }) }).pipe( + Effect.flip, + Effect.provide(child.layer), + ), + ); + expect(error).toMatchObject({ _tag: "RunSkipped", reason }); + expect(child.state.spawned).toEqual([]); + }); + + it("spawns under the budget", async () => { + const { spawned } = await runFanOut({ run: "check", items: ["a"], toInput: (item) => ({ item }) }); + expect(spawned).toHaveLength(1); + }); +}); diff --git a/packages/core/src/primitives/fan-out.ts b/packages/core/src/primitives/fan-out.ts index ebf30b9f..8732058a 100644 --- a/packages/core/src/primitives/fan-out.ts +++ b/packages/core/src/primitives/fan-out.ts @@ -20,10 +20,14 @@ // checkpoint; the deterministic child instance ids make the underlying // `create` idempotent on replay regardless. // +// Before it spawns anything, `fanOut` reads the deploy's spend through the +// capability's `spendGate`: at or over the monthly budget it fails +// `RunSkipped` and spawns no child (specs/adr/0006-spend-guard.md). +// // Spec: specs/03-dsl.md § Primitives + § spawnChildRun. import { Effect } from "effect"; -import type { ChildSpawnFailed } from "../errors"; +import type { ChildSpawnFailed, RunSkipped } from "../errors"; import { type ChildRunHandle, ChildRuns, spawnChildRun } from "../services/child-runs"; import type { Shard } from "./sharded"; @@ -65,9 +69,10 @@ export const fanOut = (opts: { readonly toInstanceId?: (item: T, shard: Shard) => string; /** Max concurrent spawns. Default: all at once (spawn is a cheap create). */ readonly concurrency?: number; -}): Effect.Effect => { +}): Effect.Effect => { const total = opts.items.length; - return Effect.forEach( + if (total === 0) return Effect.succeed([]); + const spawnAll = Effect.forEach( opts.items.map((item, i): FanOutShard => ({ item, shard: { index: i + 1, total }, @@ -78,6 +83,7 @@ export const fanOut = (opts: { input: opts.toInput(item, shard), ...(opts.toInstanceId !== undefined ? { instanceId: opts.toInstanceId(item, shard) } : {}), }), - { concurrency: opts.concurrency ?? (total === 0 ? 1 : total) }, + { concurrency: opts.concurrency ?? total }, ); + return Effect.flatMap(ChildRuns, (c) => c.spendGate(opts.run)).pipe(Effect.andThen(spawnAll)); }; diff --git a/packages/core/src/run-admission.test.ts b/packages/core/src/run-admission.test.ts index a66fd92e..50049501 100644 --- a/packages/core/src/run-admission.test.ts +++ b/packages/core/src/run-admission.test.ts @@ -8,6 +8,11 @@ import { describe, expect, it } from "vitest"; import { admissionAcquireAttempts, + admissionJitterMs, + admissionPollSleepMs, + ADMISSION_JITTER_MAX_MS, + ADMISSION_POLL_MAX_MS, + ADMISSION_POLL_MIN_MS, type AdmissionObservation, decideAdmission, } from "./run-admission"; @@ -69,9 +74,20 @@ describe("decideAdmission", () => { queuedForMs: MAX_QUEUE_AGE, position: 2, poolBusy: 16, + ceilingHeld: false, }); }); + it("carries the burn ceiling hold into the timeout", () => { + const decision = decideAdmission( + { admitted: false, position: 0, poolBusy: 3, ceilingHeld: true }, + ENQUEUED_AT, + ENQUEUED_AT + MAX_QUEUE_AGE, + MAX_QUEUE_AGE, + ); + expect(decision).toMatchObject({ _kind: "timeout", ceilingHeld: true }); + }); + it("still waits one tick before the ceiling", () => { const decision = decideAdmission( observed(), @@ -104,3 +120,58 @@ describe("admissionAcquireAttempts", () => { expect(admissionAcquireAttempts(1_000, 0)).toBeGreaterThanOrEqual(1); }); }); + +describe("admissionPollSleepMs — min(60 s, 20 s × ceil(position / cap))", () => { + it("keeps the head of the queue on the 20-second poll", () => { + expect(admissionPollSleepMs(0, 16)).toBe(20_000); + expect(admissionPollSleepMs(16, 16)).toBe(20_000); + }); + + it("scales with the slots between the waiter and the head", () => { + expect(admissionPollSleepMs(17, 16)).toBe(40_000); + expect(admissionPollSleepMs(32, 16)).toBe(40_000); + expect(admissionPollSleepMs(33, 16)).toBe(60_000); + expect(admissionPollSleepMs(9, 8)).toBe(40_000); + }); + + it("never sleeps longer than 60 seconds", () => { + expect(admissionPollSleepMs(500, 16)).toBe(ADMISSION_POLL_MAX_MS); + expect(admissionPollSleepMs(500, 1)).toBe(ADMISSION_POLL_MAX_MS); + }); + + it("shortens a long sleep to the time left, never below the 20-second poll", () => { + expect(admissionPollSleepMs(500, 16, 45_000)).toBe(45_000); + expect(admissionPollSleepMs(500, 16, 5_000)).toBe(ADMISSION_POLL_MIN_MS); + expect(admissionPollSleepMs(0, 16, -1_000)).toBe(ADMISSION_POLL_MIN_MS); + }); + + it("treats a zero cap as one slot", () => { + expect(admissionPollSleepMs(2, 0)).toBe(40_000); + }); + + it("covers the queue age within the attempts derived from the shortest sleep", () => { + // Every sleep is ≥ the shortest, so attempts × shortest ≥ the queue age. + const attempts = admissionAcquireAttempts(MAX_QUEUE_AGE, ADMISSION_POLL_MIN_MS); + expect(attempts * ADMISSION_POLL_MIN_MS).toBeGreaterThanOrEqual(MAX_QUEUE_AGE); + // A waiter far back reaches the age limit in a third of the attempts. + expect(Math.ceil(MAX_QUEUE_AGE / admissionPollSleepMs(500, 16))).toBe(attempts / 3); + }); +}); + +describe("admissionJitterMs", () => { + it("is the same for the same execution id", () => { + expect(admissionJitterMs("01ARZ3NDEKTSV4RRFFQ69G5FAA")).toBe(admissionJitterMs("01ARZ3NDEKTSV4RRFFQ69G5FAA")); + }); + + it("stays within 0..2 s and spreads a burst", () => { + const draws = Array.from({ length: 200 }, (_, i) => admissionJitterMs(`01ARZ3NDEKTSV4RRFFQ69G${String(i).padStart(4, "0")}`)); + for (const ms of draws) { + expect(ms).toBeGreaterThanOrEqual(0); + expect(ms).toBeLessThanOrEqual(ADMISSION_JITTER_MAX_MS); + expect(Number.isInteger(ms)).toBe(true); + } + expect(new Set(draws).size).toBeGreaterThan(150); + expect(Math.min(...draws)).toBeLessThan(500); + expect(Math.max(...draws)).toBeGreaterThan(1_500); + }); +}); diff --git a/packages/core/src/run-admission.ts b/packages/core/src/run-admission.ts index e35abfd5..b643264c 100644 --- a/packages/core/src/run-admission.ts +++ b/packages/core/src/run-admission.ts @@ -31,6 +31,17 @@ /** The container pool a run is admitted into — one semaphore per pool. */ export type AdmissionPool = "lean" | "browser" | "agent" | "release"; +/** + * The admission class a run waits in (ADR-0009 § Interactive admission class). + * A claim admits the oldest live `interactive` waiter in a pool before any + * `batch` waiter; FIFO order and the queue position apply within a class, + * interactive ahead of batch. A run that declares no class is `batch`. + */ +export type AdmissionClass = "interactive" | "batch"; + +/** The class of a run that declares none. */ +export const ADMISSION_CLASS_DEFAULT: AdmissionClass = "batch"; + /** * What one claim attempt observed, read atomically by the runtime layer: * whether OUR atomic claim landed, and — when it did not — where we stand. @@ -44,6 +55,12 @@ export type AdmissionObservation = { readonly position: number; /** Live admitted rows in our pool (slots currently in use). */ readonly poolBusy: number; + /** + * The burn ceiling, not the pool cap, held the claim back: our pool had a + * free slot, but the admitted hourly rates plus ours exceed the ceiling + * (ADR-0006 § Burn ceiling). Absent = false. + */ + readonly ceilingHeld?: boolean; }; /** @@ -69,6 +86,8 @@ export type AdmissionDecision = readonly queuedForMs: number; readonly position: number; readonly poolBusy: number; + /** The final attempt saw the burn ceiling, not the pool cap, holding it. */ + readonly ceilingHeld: boolean; }; /** @@ -96,6 +115,7 @@ export const decideAdmission = ( queuedForMs, position: observed.position, poolBusy: observed.poolBusy, + ceilingHeld: observed.ceilingHeld === true, }; } return { @@ -112,6 +132,59 @@ export const decideAdmission = ( * read, so the loop stays replay-deterministic across a Workflow checkpoint * resume — exactly the bound `leaseAcquireAttempts` and `waitForChildren` * use for their poll loops. + * + * With position-scaled sleeps ({@link admissionPollSleepMs}) pass the SHORTEST + * sleep: every sleep is at least that long, so the count covers the whole + * queue age, and the age read inside each claim step (`decideAdmission`) ends + * a slower-polling loop at the deadline in fewer attempts. The count is the + * ceiling on steps, the age is the ceiling on time. */ export const admissionAcquireAttempts = (maxQueueAgeMs: number, pollEveryMs: number): number => Math.max(1, Math.ceil(maxQueueAgeMs / Math.max(1, pollEveryMs))); + +/** The head-of-queue poll interval: the shortest sleep between claims. */ +export const ADMISSION_POLL_MIN_MS = 20_000; + +/** The longest sleep between claims, however far back a waiter stands. */ +export const ADMISSION_POLL_MAX_MS = 60_000; + +/** + * The sleep before a waiter's next claim (ADR-0008 § Position-scaled admission + * polls): `min(60 s, 20 s × ceil(position / cap))`, with the head (position 0) + * on the 20-second poll. A waiter moves at most one place per released slot, + * so polling far from the head admits nothing sooner. + * + * `position` is the one observed inside the claim step, so a replay sleeps the + * same length. `remainingMs` (time left before the dispatch-age ceiling, from + * the same step's clock) shortens a long sleep so the final claim lands at the + * deadline instead of up to a minute past it; the sleep never drops below the + * 20-second poll. + */ +export const admissionPollSleepMs = ( + position: number, + cap: number, + remainingMs: number = Number.POSITIVE_INFINITY, +): number => { + const slotsBack = Math.max(1, Math.ceil(Math.max(0, position) / Math.max(1, cap))); + const scaled = Math.min(ADMISSION_POLL_MAX_MS, ADMISSION_POLL_MIN_MS * slotsBack); + return Math.max(ADMISSION_POLL_MIN_MS, Math.min(scaled, remainingMs)); +}; + +/** The longest jitter an admitted run waits before its container starts. */ +export const ADMISSION_JITTER_MAX_MS = 2_000; + +/** + * The pause between an admitting claim and the container start (ADR-0009 § + * Pools sized under the spend guard), `0..ADMISSION_JITTER_MAX_MS` ms. A burst + * of admissions spreads over two seconds instead of hitting the container + * start-rate limit at once. Derived from the execution id (FNV-1a), so a + * replay waits the same length. + */ +export const admissionJitterMs = (executionId: string): number => { + let hash = 0x811c9dc5; + for (let i = 0; i < executionId.length; i++) { + hash ^= executionId.charCodeAt(i); + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash % (ADMISSION_JITTER_MAX_MS + 1); +}; diff --git a/packages/core/src/services/child-runs.ts b/packages/core/src/services/child-runs.ts index 3179db15..f8f7dd99 100644 --- a/packages/core/src/services/child-runs.ts +++ b/packages/core/src/services/child-runs.ts @@ -23,7 +23,7 @@ // § Long-running test handling + § Platform limits. import { Context, Effect } from "effect"; -import type { ChildSpawnFailed } from "../errors"; +import type { ChildSpawnFailed, RunSkipped } from "../errors"; /** * The result of a spawn. `executionId` and `instanceId` are equal — the @@ -115,6 +115,14 @@ export const isTerminalChildStatus = (status: ChildRunStatus): boolean => TERMIN /** The service contract a runtime Layer implements. */ export interface ChildRunsService { readonly spawn: (opts: SpawnChildRunOpts) => Effect.Effect; + /** + * Read the deploy's month-to-date spend before a fan-out spawns children of + * `run` (ADR-0006 § Monthly budget). At or over the monthly budget it fails + * `RunSkipped` with the budget reason, and the parent spawns none; an + * unreadable spend fails `ChildSpawnFailed` rather than reading as open. + * Call it inside the spawn step, so a replay reads the checkpointed answer. + */ + readonly spendGate: (run: string) => Effect.Effect; /** * Release the parent's global admission slot after it has finished its own * sandbox work. A parent waiting on children must not occupy a slot that a diff --git a/packages/core/src/spend.test.ts b/packages/core/src/spend.test.ts new file mode 100644 index 00000000..02c35881 --- /dev/null +++ b/packages/core/src/spend.test.ts @@ -0,0 +1,149 @@ +// Spend-guard math — the limits a deploy resolves, the month window, the +// budget decision and the burn-ceiling fit (specs/adr/0006-spend-guard.md). + +import { describe, expect, it } from "vitest"; +import { containerMicroUsdPerHour } from "./cost"; +import { + decideBudget, + fitsBurnCeiling, + monthToDateMicroUsd, + parseUsdLimit, + resolveSpendLimits, + spendCeilingReason, + spendMonth, + type SpendLimits, +} from "./spend"; + +const LIMITS: SpendLimits = { + budgetMicroUsd: 90_000_000, + burnMaxMicroUsdPerHour: 1_500_000, + overheadFactor: 1.1, + warnRatio: 0.8, +}; + +describe("resolveSpendLimits", () => { + it("caps a deploy that sets nothing at 50 USD a month and 1.50 USD an hour", () => { + expect(resolveSpendLimits({})).toEqual({ ...LIMITS, budgetMicroUsd: 50_000_000 }); + }); + + it("reads the vars, and a CONFIG_KV override wins over them", () => { + const vars = { SPEND_BUDGET_MONTHLY_USD: "200", SPEND_BURN_MAX_USD_PER_HOUR: "3" }; + expect(resolveSpendLimits(vars)).toMatchObject({ + budgetMicroUsd: 200_000_000, + burnMaxMicroUsdPerHour: 3_000_000, + }); + expect( + resolveSpendLimits(vars, { budgetMonthlyUsd: "250", burnMaxUsdPerHour: null }), + ).toMatchObject({ budgetMicroUsd: 250_000_000, burnMaxMicroUsdPerHour: 3_000_000 }); + }); + + it("disables a limit on `unlimited`, from either source", () => { + expect(resolveSpendLimits({ SPEND_BUDGET_MONTHLY_USD: "unlimited" }).budgetMicroUsd).toBeNull(); + expect( + resolveSpendLimits({}, { burnMaxUsdPerHour: " Unlimited " }).burnMaxMicroUsdPerHour, + ).toBeNull(); + }); + + it("degrades a typo to the default, never to no limit", () => { + expect(parseUsdLimit("ninety", 90)).toBe(90_000_000); + expect(parseUsdLimit("-5", 90)).toBe(90_000_000); + expect(parseUsdLimit("", 90)).toBe(90_000_000); + expect(resolveSpendLimits({ SPEND_OVERHEAD_FACTOR: "0.5", SPEND_WARN_RATIO: "2" })).toMatchObject({ + overheadFactor: 1.1, + warnRatio: 0.8, + }); + }); + + it("treats a blank override as absent", () => { + expect( + resolveSpendLimits({ SPEND_BUDGET_MONTHLY_USD: "40" }, { budgetMonthlyUsd: " " }).budgetMicroUsd, + ).toBe(40_000_000); + }); +}); + +describe("spendMonth", () => { + it("is the UTC calendar month and resets at the next one", () => { + const month = spendMonth(Date.UTC(2026, 9, 22, 13, 5)); + expect(month).toEqual({ + key: "2026-10", + startMs: Date.UTC(2026, 9, 1), + resetsAtMs: Date.UTC(2026, 10, 1), + }); + expect(spendMonth(Date.UTC(2026, 11, 31, 23, 59)).resetsAtMs).toBe(Date.UTC(2027, 0, 1)); + }); +}); + +describe("monthToDateMicroUsd", () => { + it("adds finished and in-flight spend, times the overhead factor", () => { + expect( + monthToDateMicroUsd( + { finishedMicroUsd: 10_000_000, inFlightMicroUsd: 2_000_000, admittedMicroUsdPerHour: 0, admittedCount: 0 }, + 1.1, + ), + ).toBe(13_200_000); + }); +}); + +describe("decideBudget", () => { + it("stays open under the budget and warns past the warn ratio", () => { + expect(decideBudget(10_000_000, LIMITS)).toEqual({ _kind: "open", warn: false }); + expect(decideBudget(72_000_000, LIMITS)).toEqual({ _kind: "open", warn: true }); + }); + + it("closes at the budget with the skip reason", () => { + expect(decideBudget(90_000_000, LIMITS)).toEqual({ + _kind: "closed", + reason: "spend budget reached: $90.00 of $90.00 this month", + }); + expect(decideBudget(91_234_567, LIMITS)).toMatchObject({ + reason: "spend budget reached: $91.23 of $90.00 this month", + }); + }); + + it("never closes an unlimited budget", () => { + expect(decideBudget(10 ** 12, { ...LIMITS, budgetMicroUsd: null })).toEqual({ + _kind: "open", + warn: false, + }); + }); +}); + +describe("fitsBurnCeiling", () => { + const fits = (admitted: readonly number[], own: number, ceiling: number | null = 1_500_000) => + fitsBurnCeiling({ + admittedMicroUsdPerHour: admitted.reduce((a, b) => a + b, 0), + admittedCount: admitted.length, + ownMicroUsdPerHour: own, + burnMaxMicroUsdPerHour: ceiling, + }); + + it("admits three standard-4 containers, or eleven standard-2, under the default ceiling", () => { + const s4 = containerMicroUsdPerHour("standard-4"); + const s2 = containerMicroUsdPerHour("standard-2"); + expect(fits([s4, s4], s4)).toBe(true); + expect(fits([s4, s4, s4], s4)).toBe(false); + expect(fits(Array(10).fill(s2), s2)).toBe(true); + expect(fits(Array(11).fill(s2), s2)).toBe(false); + }); + + it("admits a claimant whose own rate exceeds the ceiling into an empty set only", () => { + expect(fits([], 2_000_000)).toBe(true); + expect(fits([100_000], 2_000_000)).toBe(false); + }); + + it("always fits with no ceiling", () => { + expect(fits([10_000_000], 10_000_000, null)).toBe(true); + }); +}); + +describe("spendCeilingReason", () => { + it("names the burn and the ceiling", () => { + expect( + spendCeilingReason({ + admittedMicroUsdPerHour: 1_203_120, + ownMicroUsdPerHour: 401_040, + burnMaxMicroUsdPerHour: 1_500_000, + }), + ).toBe("spend ceiling: admitted runs burn $1.20/h of the $1.50/h ceiling, this run needs $0.40/h"); + }); +}); diff --git a/packages/core/src/spend.ts b/packages/core/src/spend.ts new file mode 100644 index 00000000..68d44ef8 --- /dev/null +++ b/packages/core/src/spend.ts @@ -0,0 +1,198 @@ +// @fractalboxdev/flare-dispatch-core — spend guard: the pure budget and burn-ceiling math. +// +// Cloudflare has no spend cap for Containers, so admission is where a deploy +// caps its own spend: every run passes the admission semaphore before it +// starts a container. Two limits bind there: +// +// * a MONTHLY BUDGET — month-to-date spend at or over it closes admission, +// and the run ends `RunSkipped` (a neutral check, a `skipped` row); +// * a BURN CEILING — the summed hourly rate of every admitted execution, +// across all pools, stays at or under it. The claim SQL enforces it +// (runtime-cf run-admission-d1.ts); `fitsBurnCeiling` is its pure mirror. +// +// Money is integer micro-USD throughout, like the cost engine (cost.ts). This +// module holds no I/O and no clock: the runtime reads D1 and CONFIG_KV inside +// a durable step and hands the readings here. +// +// Spec: specs/adr/0006-spend-guard.md. + +import { MICRO_USD_PER_USD } from "./cost"; + +/** The monthly budget a deploy that sets nothing gets, in USD. */ +export const SPEND_BUDGET_MONTHLY_USD_DEFAULT = 50; +/** The burn ceiling a deploy that sets nothing gets, in USD per hour. */ +export const SPEND_BURN_MAX_USD_PER_HOUR_DEFAULT = 1.5; +/** Multiplier for Workflow steps and Durable Objects on top of container + model spend. */ +export const SPEND_OVERHEAD_FACTOR_DEFAULT = 1.1; +/** Fraction of the budget whose crossing sends the month's one warning. */ +export const SPEND_WARN_RATIO_DEFAULT = 0.8; + +/** The value that disables a limit, in a var or a `CONFIG_KV` override. */ +export const SPEND_UNLIMITED = "unlimited"; + +/** The spend-guard settings an admission decides with. `null` = unlimited. */ +export type SpendLimits = { + readonly budgetMicroUsd: number | null; + readonly burnMaxMicroUsdPerHour: number | null; + readonly overheadFactor: number; + readonly warnRatio: number; +}; + +/** The raw spend settings: wrangler vars, each overridable from `CONFIG_KV`. */ +export type SpendVars = { + readonly SPEND_BUDGET_MONTHLY_USD?: string | undefined; + readonly SPEND_BURN_MAX_USD_PER_HOUR?: string | undefined; + readonly SPEND_OVERHEAD_FACTOR?: string | undefined; + readonly SPEND_WARN_RATIO?: string | undefined; +}; + +/** The `CONFIG_KV` overrides an operator writes to change a limit without a deploy. */ +export type SpendOverrides = { + readonly budgetMonthlyUsd?: string | null | undefined; + readonly burnMaxUsdPerHour?: string | null | undefined; +}; + +const toMicro = (usd: number): number => Math.round(usd * MICRO_USD_PER_USD); + +/** A present, non-blank override wins; otherwise the var. */ +const pick = (override: string | null | undefined, value: string | undefined): string | undefined => + override !== null && override !== undefined && override.trim() !== "" ? override : value; + +/** + * Parse a USD limit into micro-USD. `unlimited` → `null`; a finite number ≥ 0 + * → its micro-USD value; absent, blank or garbage → `fallbackUsd`. A typo + * degrades to the default limit, never to no limit. + */ +export const parseUsdLimit = (raw: string | null | undefined, fallbackUsd: number): number | null => { + const text = raw?.trim(); + if (text === undefined || text === "") return toMicro(fallbackUsd); + if (text.toLowerCase() === SPEND_UNLIMITED) return null; + const usd = Number(text); + return Number.isFinite(usd) && usd >= 0 ? toMicro(usd) : toMicro(fallbackUsd); +}; + +/** Parse a bounded ratio; outside `[min, max]` or unparsable → `fallback`. */ +const parseRatio = (raw: string | undefined, fallback: number, min: number, max: number): number => { + const text = raw?.trim(); + if (text === undefined || text === "") return fallback; + const value = Number(text); + return Number.isFinite(value) && value >= min && value <= max ? value : fallback; +}; + +/** + * Resolve the limits: a present `CONFIG_KV` override wins over the var, and + * the var over the code default. The defaults live here rather than in the + * committed wrangler.jsonc because an operator overlay replaces that file. + */ +export const resolveSpendLimits = (vars: SpendVars, overrides: SpendOverrides = {}): SpendLimits => ({ + budgetMicroUsd: parseUsdLimit( + pick(overrides.budgetMonthlyUsd, vars.SPEND_BUDGET_MONTHLY_USD), + SPEND_BUDGET_MONTHLY_USD_DEFAULT, + ), + burnMaxMicroUsdPerHour: parseUsdLimit( + pick(overrides.burnMaxUsdPerHour, vars.SPEND_BURN_MAX_USD_PER_HOUR), + SPEND_BURN_MAX_USD_PER_HOUR_DEFAULT, + ), + overheadFactor: parseRatio(vars.SPEND_OVERHEAD_FACTOR, SPEND_OVERHEAD_FACTOR_DEFAULT, 1, 10), + warnRatio: parseRatio(vars.SPEND_WARN_RATIO, SPEND_WARN_RATIO_DEFAULT, Number.MIN_VALUE, 1), +}); + +/** The current UTC calendar month: its bounds and its `YYYY-MM` key. */ +export type SpendMonth = { + readonly key: string; + readonly startMs: number; + /** When the month resets — the first instant of the next UTC month. */ + readonly resetsAtMs: number; +}; + +/** The UTC calendar month containing `nowMs`. */ +export const spendMonth = (nowMs: number): SpendMonth => { + const now = new Date(nowMs); + const year = now.getUTCFullYear(); + const month = now.getUTCMonth(); + return { + key: `${year}-${String(month + 1).padStart(2, "0")}`, + startMs: Date.UTC(year, month, 1), + resetsAtMs: Date.UTC(year, month + 1, 1), + }; +}; + +/** + * What one spend read observed in D1 (runtime-cf spend-d1.ts), before the + * overhead factor: + * + * * `finishedMicroUsd` — `cost_micro_usd` of executions admitted this month; + * * `inFlightMicroUsd` — each admitted, unfinished execution's hourly rate × + * the time since its admission; + * * `admittedMicroUsdPerHour` / `admittedCount` — the live admitted set the + * burn ceiling sums, across every pool. + */ +export type SpendReading = { + readonly finishedMicroUsd: number; + readonly inFlightMicroUsd: number; + readonly admittedMicroUsdPerHour: number; + readonly admittedCount: number; +}; + +/** Month-to-date spend: finished + in-flight, times the overhead factor. */ +export const monthToDateMicroUsd = (reading: SpendReading, overheadFactor: number): number => + Math.round((reading.finishedMicroUsd + reading.inFlightMicroUsd) * overheadFactor); + +/** `$12.34` — two decimals, the precision a budget line reads at. */ +export const formatUsd = (microUsd: number): string => `$${(microUsd / MICRO_USD_PER_USD).toFixed(2)}`; + +/** + * The monthly-budget decision for one admission or spawn. + * + * - `open` — under budget (or unlimited); `warn` is true once spend sits at + * or over `warnRatio` of the budget. + * - `closed` — at or over budget: the run skips with `reason`. + */ +export type BudgetDecision = + | { readonly _kind: "open"; readonly warn: boolean } + | { readonly _kind: "closed"; readonly reason: string }; + +/** Decide the monthly budget from month-to-date spend. Pure. */ +export const decideBudget = (spendMicroUsd: number, limits: SpendLimits): BudgetDecision => { + const budget = limits.budgetMicroUsd; + if (budget === null) return { _kind: "open", warn: false }; + if (spendMicroUsd >= budget) { + return { + _kind: "closed", + reason: `spend budget reached: ${formatUsd(spendMicroUsd)} of ${formatUsd(budget)} this month`, + }; + } + return { _kind: "open", warn: spendMicroUsd >= budget * limits.warnRatio }; +}; + +/** + * Whether a claimant at `ownMicroUsdPerHour` fits under the burn ceiling — the + * pure mirror of the claim SQL's ceiling clause. The admitted rates plus the + * claimant's own must fit; a claimant whose own rate alone exceeds the + * ceiling fits only an empty admitted set, so it never waits forever. + */ +export const fitsBurnCeiling = (opts: { + readonly admittedMicroUsdPerHour: number; + readonly admittedCount: number; + readonly ownMicroUsdPerHour: number; + readonly burnMaxMicroUsdPerHour: number | null; +}): boolean => + opts.burnMaxMicroUsdPerHour === null || + opts.admittedCount === 0 || + opts.admittedMicroUsdPerHour + opts.ownMicroUsdPerHour <= opts.burnMaxMicroUsdPerHour; + +/** The `RunSkipped` reason for a run the burn ceiling held in the queue past its age. */ +export const spendCeilingReason = (opts: { + readonly admittedMicroUsdPerHour: number; + readonly ownMicroUsdPerHour: number; + readonly burnMaxMicroUsdPerHour: number; +}): string => + `spend ceiling: admitted runs burn ${formatUsd(opts.admittedMicroUsdPerHour)}/h of the ` + + `${formatUsd(opts.burnMaxMicroUsdPerHour)}/h ceiling, this run needs ${formatUsd(opts.ownMicroUsdPerHour)}/h`; + +/** The `CONFIG_KV` key that overrides the monthly budget (USD, or `unlimited`). */ +export const SPEND_BUDGET_KV_KEY = "spend.budgetMonthlyUsd"; +/** The `CONFIG_KV` key that overrides the burn ceiling (USD per hour, or `unlimited`). */ +export const SPEND_BURN_KV_KEY = "spend.burnMaxUsdPerHour"; +/** The `CONFIG_KV` marker that a month's warning went out. */ +export const spendWarnedKvKey = (monthKey: string): string => `spend.warned:${monthKey}`; diff --git a/packages/runtime-cf/package.json b/packages/runtime-cf/package.json index 8d6f6579..94d47d85 100644 --- a/packages/runtime-cf/package.json +++ b/packages/runtime-cf/package.json @@ -10,6 +10,7 @@ "./model-gateway": "./src/model-gateway-cf.ts", "./log-names": "./src/log-names.ts", "./admission-cap": "./src/run-admission-d1.ts", + "./spend": "./src/spend-d1.ts", "./run-supersede": "./src/run-supersede-d1.ts" }, "scripts": { diff --git a/packages/runtime-cf/src/child-runs-cf.ts b/packages/runtime-cf/src/child-runs-cf.ts index 759ad07a..b538cad5 100644 --- a/packages/runtime-cf/src/child-runs-cf.ts +++ b/packages/runtime-cf/src/child-runs-cf.ts @@ -35,6 +35,7 @@ import { ChildSpawnFailed, type ChildStatusRecord, isTerminalChildStatus, + RunSkipped, } from "@fractalboxdev/flare-dispatch-core"; import { makeRunAdmissionD1 } from "./run-admission-d1"; @@ -85,6 +86,14 @@ export type ChildRunsLiveConfig = { * Workflow has no request to infer it from). Absent → relative paths. */ readonly origin?: string; + /** + * The deploy's monthly-budget read (ADR-0006): resolves the `RunSkipped` + * reason when month-to-date spend is at or over the budget, `undefined` + * while it is open; rejects when spend is unreadable. The dispatcher wires + * it over D1 + `CONFIG_KV`. Absent = a runtime with no spend guard, the + * same as an `unlimited` budget. + */ + readonly spendGate?: () => Promise; }; /** Map a raw `executions.status` cell to the capability's `ChildRunStatus`. */ @@ -154,7 +163,19 @@ export const deriveChildInstanceId = (opts: { /** Build the live `ChildRuns` Layer bound to a CF `Workflow` binding. */ export const makeChildRunsLive = (cfg: ChildRunsLiveConfig): Layer.Layer => { + const spendGate = cfg.spendGate; const service: ChildRunsService = { + spendGate: (run) => + spendGate === undefined + ? Effect.void + : Effect.tryPromise({ + try: spendGate, + catch: (cause) => new ChildSpawnFailed({ run, instanceId: "spend-gate", cause }), + }).pipe( + Effect.flatMap((closed) => + closed === undefined ? Effect.void : Effect.fail(new RunSkipped({ reason: closed })), + ), + ), handoffAdmission: () => makeRunAdmissionD1(cfg.db).release(cfg.parentExecutionId).pipe(Effect.orDie), spawn: ({ run, input, instanceId }) => { diff --git a/packages/runtime-cf/src/deferred.ts b/packages/runtime-cf/src/deferred.ts index 11cf1a51..b9121cf8 100644 --- a/packages/runtime-cf/src/deferred.ts +++ b/packages/runtime-cf/src/deferred.ts @@ -79,6 +79,8 @@ export const ChildRunsDeferred: Layer.Layer = Layer.succeed( Effect.die(`spawnChildRun: no RUNS_WORKFLOW binding on this runtime (run="${run}")`), handoffAdmission: () => Effect.die("handoffChildAdmission: no RUNS_WORKFLOW binding on this runtime"), + spendGate: (run) => + Effect.die(`fanOut: no RUNS_WORKFLOW binding on this runtime (run="${run}")`), poll: () => Effect.die("waitForChildren: no RUNS_WORKFLOW binding on this runtime"), }))(), ); diff --git a/packages/runtime-cf/src/execution-cost.test.ts b/packages/runtime-cf/src/execution-cost.test.ts new file mode 100644 index 00000000..6f62659e --- /dev/null +++ b/packages/runtime-cf/src/execution-cost.test.ts @@ -0,0 +1,49 @@ +// The sandbox → container class → instance type table the meter and the burn +// ceiling price runs from, checked against the committed wrangler.jsonc: a +// container class resized there fails here until the table follows +// (specs/adr/0006-spend-guard.md § Meter). + +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; +import { containerMicroUsdPerHour, type SandboxImage } from "@fractalboxdev/flare-dispatch-core"; +import { hourlyRateForSandboxImage, instanceForSandboxImage, SANDBOX_CONTAINER } from "./execution-cost"; + +/** The committed `wrangler.jsonc`, comments and trailing commas stripped. */ +const wrangler = (() => { + const raw = readFileSync(new URL("../../../wrangler.jsonc", import.meta.url), "utf8"); + return JSON.parse( + raw + .replace(/\/\*[\s\S]*?\*\//g, "") + .replace(/(^|[^:])\/\/.*$/gm, "$1") + .replace(/,(\s*[}\]])/g, "$1"), + ) as { + containers: { class_name: string; instance_type: string }[]; + durable_objects: { bindings: { name: string; class_name: string }[] }; + }; +})(); + +/** The Container binding each sandbox image routes to (sandbox-routing.ts). */ +const IMAGE_BINDING: Record = { + lean: "RUNS_SANDBOX", + browser: "RUNS_SANDBOX_BROWSER", + agent: "RUNS_SANDBOX_AGENT", + release: "RUNS_SANDBOX_RELEASE", +}; + +describe("SANDBOX_CONTAINER — the one instance table", () => { + it("matches each image's container class and instance_type in wrangler.jsonc", () => { + for (const image of Object.keys(SANDBOX_CONTAINER) as SandboxImage[]) { + const binding = wrangler.durable_objects.bindings.find((b) => b.name === IMAGE_BINDING[image]); + expect(binding?.class_name, `${image} binding`).toBe(SANDBOX_CONTAINER[image].className); + const container = wrangler.containers.find((c) => c.class_name === SANDBOX_CONTAINER[image].className); + expect(container?.instance_type, `${image} container`).toBe(SANDBOX_CONTAINER[image].instance); + } + }); + + it("prices the lean image on its standard-4 class, and an unset image as lean", () => { + expect(instanceForSandboxImage("lean")).toBe("standard-4"); + expect(instanceForSandboxImage(undefined)).toBe("standard-4"); + expect(instanceForSandboxImage("browser")).toBe("standard-2"); + expect(hourlyRateForSandboxImage("agent")).toBe(containerMicroUsdPerHour("standard-3")); + }); +}); diff --git a/packages/runtime-cf/src/execution-cost.ts b/packages/runtime-cf/src/execution-cost.ts index caa09b98..1a53752c 100644 --- a/packages/runtime-cf/src/execution-cost.ts +++ b/packages/runtime-cf/src/execution-cost.ts @@ -5,32 +5,60 @@ // // * METERED model tokens — SUMmed from `execution_model_usage` (written per // call by the live `modelGateway` wrapper, model-gateway-cf.ts); -// * MODELED container compute — `instance vCPU/mem rate × active seconds`, -// where active seconds is the execution wall-time (an upper-bound proxy: -// it includes model round-trips + any in-body waits; refining this to the -// true sandbox boot→destroy window is a follow-up). +// * MODELED container compute — the list rate of the run's container class, +// priced on provisioned vCPU, memory and disk, × container-active seconds, +// which run from admission (`executions.admitted_at`) to completion. Queue +// wait costs nothing; an execution never admitted has no container time. // // via the pure `@fractalboxdev/flare-dispatch-core` cost engine, which decides the honest -// `metered | mixed | modeled | unmetered` basis. Best-effort: a cost-rollup -// failure logs a warning and never flips the run's verdict. +// `metered | mixed | modeled | unmetered` basis. Each row it writes carries +// `cost_meter = 'admitted-active'`; rows priced on wall time before this rule +// carry `'modeled-wall'` (migration 0011). Best-effort: a cost-rollup failure +// logs a warning and never flips the run's verdict. // -// Spec: specs/06-cost.md, infra/migrations/0005_execution_cost.sql. +// Spec: specs/06-cost.md, specs/adr/0006-spend-guard.md § Meter, +// infra/migrations/0005_execution_cost.sql, 0011_spend_guard.sql. import { Effect } from "effect"; import { type InstanceType, type SandboxImage, INSTANCE_SPECS, + containerMicroUsdPerHour, estimateExecutionCost, } from "@fractalboxdev/flare-dispatch-core"; /** - * The container instance type a run's `sandboxImage` lands on. The Durable - * Object classes in wrangler.jsonc bind `lean`/`browser` → `standard-2` and - * `agent` → `standard-3`; an un-set image defaults to `lean`. + * The container class and instance type each sandbox image (= admission pool) + * runs on — the one table the meter, the burn ceiling and the finish path + * read. `execution-cost.test.ts` checks it against the committed + * wrangler.jsonc `containers` entries, so a class resized there fails CI until + * this table follows. + */ +export const SANDBOX_CONTAINER: Record< + SandboxImage, + { readonly className: string; readonly instance: InstanceType } +> = { + lean: { className: "RunSandbox", instance: "standard-4" }, + browser: { className: "RunSandboxBrowser", instance: "standard-2" }, + agent: { className: "RunSandboxAgent", instance: "standard-3" }, + release: { className: "RunSandboxRelease", instance: "standard-4" }, +}; + +/** + * The container instance type a run lands on. Pass the ADMISSION POOL (the + * image after the dispatcher's degrade-to-lean routing), so a run is priced on + * the container it actually consumes; an un-set image is `lean`. */ export const instanceForSandboxImage = (sandboxImage: SandboxImage | undefined): InstanceType => - sandboxImage === "agent" ? "standard-3" : sandboxImage === "release" ? "standard-4" : "standard-2"; + SANDBOX_CONTAINER[sandboxImage ?? "lean"].instance; + +/** A sandbox image's list rate per hour, in micro-USD — what the burn ceiling sums. */ +export const hourlyRateForSandboxImage = (sandboxImage: SandboxImage | undefined): number => + containerMicroUsdPerHour(instanceForSandboxImage(sandboxImage)); + +/** The `executions.cost_meter` label for a row priced by this rule. */ +export const COST_METER_ADMITTED_ACTIVE = "admitted-active"; type UsageRow = { readonly model: string; @@ -40,13 +68,13 @@ type UsageRow = { }; /** - * Compute and persist this execution's cost rollup. Reads its own wall-time and + * Compute and persist this execution's cost rollup. Reads its admitted window and * model-usage rows from D1, models container compute from the instance type, and * writes the `executions` cost columns. Returns `void`; all failure is absorbed. * * @param db D1 binding (`env.RUNS_METADATA`). * @param executionId the row to roll up. - * @param instance the container instance type (`instanceForSandboxImage`). + * @param instance the container instance type (`instanceForSandboxImage(pool)`). */ export const recordExecutionCost = (opts: { readonly db: D1Database; @@ -56,16 +84,17 @@ export const recordExecutionCost = (opts: { Effect.tryPromise(async () => { const { db, executionId, instance } = opts; - // Wall-time → modeled container-active seconds (upper-bound proxy). + // Admission → completion is the container-active window; queue wait and + // a run never admitted cost no container time. const exec = await db - .prepare(`SELECT started_at, completed_at FROM executions WHERE id = ?`) + .prepare(`SELECT admitted_at, completed_at FROM executions WHERE id = ?`) .bind(executionId) - .first<{ started_at: number | null; completed_at: number | null }>(); - const startedAt = exec?.started_at ?? null; + .first<{ admitted_at: number | null; completed_at: number | null }>(); + const admittedAt = exec?.admitted_at ?? null; const completedAt = exec?.completed_at ?? null; const activeSeconds = - startedAt !== null && completedAt !== null && completedAt > startedAt - ? (completedAt - startedAt) / 1000 + admittedAt !== null && completedAt !== null && completedAt > admittedAt + ? (completedAt - admittedAt) / 1000 : 0; // SUM model token usage; pick the dominant model for the rate + the column. @@ -109,13 +138,14 @@ export const recordExecutionCost = (opts: { await db .prepare( `UPDATE executions - SET cost_micro_usd = ?, cost_basis = ?, input_tokens = ?, + SET cost_micro_usd = ?, cost_basis = ?, cost_meter = ?, input_tokens = ?, output_tokens = ?, vcpu_seconds = ?, model = ? WHERE id = ?`, ) .bind( cost.totalMicroUsd, cost.basis, + COST_METER_ADMITTED_ACTIVE, rows.length > 0 ? inTok : null, rows.length > 0 ? outTok : null, vcpuSeconds, diff --git a/packages/runtime-cf/src/execution-cost.workers.test.ts b/packages/runtime-cf/src/execution-cost.workers.test.ts index 234e87c6..115b43de 100644 --- a/packages/runtime-cf/src/execution-cost.workers.test.ts +++ b/packages/runtime-cf/src/execution-cost.workers.test.ts @@ -12,16 +12,26 @@ import { makeTestBindings, type TestBindings } from "./test-support-workers"; const EXEC = "01TEST0000000000000000COST"; -/** Insert a finished executions row spanning `wallMs` of wall-time. */ -const seedExecution = (db: D1Database, id: string, wallMs: number) => - db +/** + * Insert a finished executions row admitted `activeMs` before it completed, + * after `queuedMs` in the admission queue (`admittedAt: null` = never admitted). + */ +const seedExecution = ( + db: D1Database, + id: string, + activeMs: number, + opts: { readonly queuedMs?: number; readonly admitted?: boolean } = {}, +) => { + const admittedAt = 1000 + (opts.queuedMs ?? 0); + return db .prepare( `INSERT INTO executions - (id, run, repo, ref, sha, status, started_at, completed_at, input_json) - VALUES (?, 'pr-review', 'owner/name', 'refs/heads/main', 'abc', 'success', 1000, ?, '{}')`, + (id, run, repo, ref, sha, status, started_at, admitted_at, completed_at, input_json) + VALUES (?, 'pr-review', 'owner/name', 'refs/heads/main', 'abc', 'success', 1000, ?, ?, '{}')`, ) - .bind(id, 1000 + wallMs) + .bind(id, opts.admitted === false ? null : admittedAt, admittedAt + activeMs) .run(); +}; const seedUsage = ( db: D1Database, @@ -50,8 +60,9 @@ describe("recordExecutionCost", () => { }); it("rolls up metered model tokens + modeled container compute as 'mixed'", async () => { - // 480s wall on standard-2 → 16_800 µ$ modeled container; sonnet 20k in / - // 2k out → 20000×3 + 2000×15 = 90_000 µ$ metered model. + // 480s admitted on standard-2 → 17_203 µ$ modeled container (vCPU + + // memory + disk); sonnet 20k in / 2k out → 20000×3 + 2000×15 = 90_000 µ$ + // metered model. await seedExecution(bindings.db, EXEC, 480_000); await seedUsage(bindings.db, EXEC, "anthropic/claude-sonnet-4-6", 20_000, 2_000, 1); @@ -68,7 +79,7 @@ describe("recordExecutionCost", () => { .first>(); expect(row).toMatchObject({ - cost_micro_usd: 106_800, // 16_800 container + 90_000 model + cost_micro_usd: 107_203, // 17_203 container + 90_000 model cost_basis: "mixed", input_tokens: 20_000, output_tokens: 2_000, @@ -92,12 +103,43 @@ describe("recordExecutionCost", () => { .bind(EXEC) .first>(); - expect(row).toMatchObject({ cost_micro_usd: 16_800, cost_basis: "modeled" }); + expect(row).toMatchObject({ cost_micro_usd: 17_203, cost_basis: "modeled" }); expect(row?.input_tokens).toBeNull(); expect(row?.output_tokens).toBeNull(); expect(row?.model).toBeNull(); }); + it("prices admission to completion: queue wait costs nothing, and the row reads admitted-active", async () => { + // 10 minutes queued, then 480 s admitted on standard-4. + await seedExecution(bindings.db, EXEC, 480_000, { queuedMs: 600_000 }); + + await Effect.runPromise( + recordExecutionCost({ db: bindings.db, executionId: EXEC, instance: "standard-4" }), + ); + + const row = await bindings.db + .prepare(`SELECT cost_micro_usd, cost_meter, vcpu_seconds FROM executions WHERE id = ?`) + .bind(EXEC) + .first<{ cost_micro_usd: number; cost_meter: string; vcpu_seconds: number }>(); + // (4 × 20 + 12 × 2.5 + 20 × 0.07) µ$/s × 480 s = 111.4 × 480. + expect(row).toMatchObject({ cost_micro_usd: 53_472, cost_meter: "admitted-active" }); + expect(row?.vcpu_seconds).toBeCloseTo(1920, 3); + }); + + it("charges no container time to an execution never admitted", async () => { + await seedExecution(bindings.db, EXEC, 480_000, { admitted: false }); + + await Effect.runPromise( + recordExecutionCost({ db: bindings.db, executionId: EXEC, instance: "standard-4" }), + ); + + const row = await bindings.db + .prepare(`SELECT cost_micro_usd FROM executions WHERE id = ?`) + .bind(EXEC) + .first<{ cost_micro_usd: number }>(); + expect(row?.cost_micro_usd).toBe(0); + }); + it("upsert-SUM accumulates a same-model fan-out (no double-count on the PK)", async () => { await seedExecution(bindings.db, EXEC, 10_000); // Two writes to the SAME (execution, model) PK — the gateway wrapper's diff --git a/packages/runtime-cf/src/index.ts b/packages/runtime-cf/src/index.ts index a2f91b29..9dcb3351 100644 --- a/packages/runtime-cf/src/index.ts +++ b/packages/runtime-cf/src/index.ts @@ -60,11 +60,14 @@ export { } from "./container-lease-d1"; export { makeRunAdmissionD1, + BURN_UNLIMITED, + type BurnCeiling, resolveAdmissionCap, resolvePoolAdmissionCap, ADMISSION_CAP_DEFAULT, RELEASE_ADMISSION_CAP_DEFAULT, AGENT_ADMISSION_CAP_DEFAULT, + BROWSER_ADMISSION_CAP_DEFAULT, ADMISSION_POLL_EVERY_MS, ADMISSION_MAX_QUEUE_AGE_MS, ADMISSION_TTL_MS, @@ -116,7 +119,20 @@ export { type WorkflowBindingLike, } from "./child-runs-cf"; export { makeModelGatewayLive, type AiBinding, type ModelUsageSink } from "./model-gateway-cf"; -export { recordExecutionCost, instanceForSandboxImage } from "./execution-cost"; +export { + recordExecutionCost, + instanceForSandboxImage, + hourlyRateForSandboxImage, + SANDBOX_CONTAINER, + COST_METER_ADMITTED_ACTIVE, +} from "./execution-cost"; +export { + readSpend, + readSpendLimits, + readSpendStatus, + warnSpendOnce, + type SpendStatus, +} from "./spend-d1"; export { BrowserDeferred, ChildRunsDeferred, diff --git a/packages/runtime-cf/src/run-admission-d1.ts b/packages/runtime-cf/src/run-admission-d1.ts index 79f03650..981e922d 100644 --- a/packages/runtime-cf/src/run-admission-d1.ts +++ b/packages/runtime-cf/src/run-admission-d1.ts @@ -26,9 +26,30 @@ // Spec: ADR #109 (global D1-backed run admission semaphore). import { Effect, Schedule } from "effect"; -import type { AdmissionObservation, AdmissionPool } from "@fractalboxdev/flare-dispatch-core"; +import { + ADMISSION_CLASS_DEFAULT, + ADMISSION_POLL_MAX_MS, + ADMISSION_POLL_MIN_MS, + type AdmissionClass, + type AdmissionObservation, + type AdmissionPool, + fitsBurnCeiling, +} from "@fractalboxdev/flare-dispatch-core"; import { LEASE_TTL_MS } from "./container-lease-d1"; +/** + * The spend-guard burn ceiling one claimant admits under (ADR-0006 § Burn + * ceiling): its own container class's list rate per hour, and the ceiling on + * the summed rate of every admitted row across all pools (`null` = unlimited). + */ +export type BurnCeiling = { + readonly hourlyRateMicroUsd: number; + readonly burnMaxMicroUsdPerHour: number | null; +}; + +/** No ceiling, no rate — the store as it behaves without a spend guard. */ +export const BURN_UNLIMITED: BurnCeiling = { hourlyRateMicroUsd: 0, burnMaxMicroUsdPerHour: null }; + /** * Default admission slots per pool — matches each container class's * `max_instances: 16` in wrangler.jsonc, so admitted runs never exceed the @@ -36,7 +57,7 @@ import { LEASE_TTL_MS } from "./container-lease-d1"; * `ADMISSION_CAP` wrangler var (see {@link resolveAdmissionCap}); this is * the fallback when the var is unset. */ -export const ADMISSION_CAP_DEFAULT = 16; +export const ADMISSION_CAP_DEFAULT = 48; /** * Parse the `ADMISSION_CAP` wrangler var (vars are always strings) into the @@ -66,26 +87,35 @@ export const RELEASE_ADMISSION_CAP_DEFAULT = 8; */ export const AGENT_ADMISSION_CAP_DEFAULT = 8; +/** + * The browser pool's slot cap when `ADMISSION_CAP_BROWSER` is unset — the + * `RunSandboxBrowser` container's `max_instances: 16` in wrangler.jsonc. + */ +export const BROWSER_ADMISSION_CAP_DEFAULT = 16; + /** The admission vars a pool's cap reads. */ export type AdmissionCapVars = { readonly ADMISSION_CAP?: string; readonly ADMISSION_CAP_AGENT?: string; + readonly ADMISSION_CAP_BROWSER?: string; readonly ADMISSION_CAP_RELEASE?: string; }; /** - * The slot cap for one pool. `ADMISSION_CAP` bounds every pool; the agent and - * release pools, whose containers are fewer, are further bounded by - * `ADMISSION_CAP_AGENT` and `ADMISSION_CAP_RELEASE` (defaults - * {@link AGENT_ADMISSION_CAP_DEFAULT}, {@link RELEASE_ADMISSION_CAP_DEFAULT}), - * so their cap is the smaller of the two. Lean and browser keep - * `ADMISSION_CAP` alone. Garbage degrades to the default, as in - * {@link resolveAdmissionCap}. + * The slot cap for one pool — an upper bound; the spend guard's burn ceiling + * sets the operating concurrency (ADR-0009 § Pools sized under the spend + * guard). `ADMISSION_CAP` bounds every pool; the agent, browser and release + * pools are further bounded by `ADMISSION_CAP_AGENT`, `ADMISSION_CAP_BROWSER` + * and `ADMISSION_CAP_RELEASE` (defaults {@link AGENT_ADMISSION_CAP_DEFAULT}, + * {@link BROWSER_ADMISSION_CAP_DEFAULT}, {@link RELEASE_ADMISSION_CAP_DEFAULT}), + * so their cap is the smaller of the two. Lean keeps `ADMISSION_CAP` alone. + * Garbage degrades to the default, as in {@link resolveAdmissionCap}. */ export const resolvePoolAdmissionCap = (pool: AdmissionPool, vars: AdmissionCapVars): number => { const shared = resolveAdmissionCap(vars.ADMISSION_CAP); const own = pool === "release" ? { raw: vars.ADMISSION_CAP_RELEASE, fallback: RELEASE_ADMISSION_CAP_DEFAULT } : pool === "agent" ? { raw: vars.ADMISSION_CAP_AGENT, fallback: AGENT_ADMISSION_CAP_DEFAULT } + : pool === "browser" ? { raw: vars.ADMISSION_CAP_BROWSER, fallback: BROWSER_ADMISSION_CAP_DEFAULT } : undefined; if (own === undefined) return shared; const parsed = own.raw === undefined ? NaN : Number.parseInt(own.raw, 10); @@ -93,12 +123,16 @@ export const resolvePoolAdmissionCap = (pool: AdmissionPool, vars: AdmissionCapV }; /** - * Delay between claim attempts while the pool is full. Each wait is a - * durable `step.sleep`, so a queued run costs nothing while it waits; 20s - * keeps the loop ≤270 claim steps + 270 sleeps over the largest run override - * — trivial against the 10k Workflow step cap (#83 precedent). + * The shortest delay between claim attempts — the head of the queue's poll. + * A waiter further back sleeps up to {@link ADMISSION_POLL_MAX_MS} + * (`admissionPollSleepMs`, core run-admission.ts). Each wait is a durable + * `step.sleep`, so a queued run costs nothing while it waits. The attempt + * count is derived from this shortest sleep, so one attempt costs three steps + * (spend check, claim, sleep): the largest run override (6 h) bounds the loop + * at 1,080 attempts, 3,240 steps, under the 10k Workflow step cap (#83 + * precedent); a waiter several slots back polls up to three times less often. */ -export const ADMISSION_POLL_EVERY_MS = 20_000; +export const ADMISSION_POLL_EVERY_MS = ADMISSION_POLL_MIN_MS; /** * Dispatch-age ceiling — a run queued longer than this fails @@ -119,11 +153,13 @@ export const ADMISSION_TTL_MS = LEASE_TTL_MS; /** * Heartbeat staleness ceiling for QUEUED rows. A waiter refreshes its - * heartbeat on every claim attempt (every `ADMISSION_POLL_EVERY_MS`), so a - * row silent for ~5 polls is a dead waiter — excluded from the FIFO no-barge - * check and the position count, so a corpse never blocks the line. + * heartbeat on every claim attempt, at most {@link ADMISSION_POLL_MAX_MS} + * apart, so a row silent for three of the longest sleeps is a dead waiter — + * excluded from the FIFO no-barge check and the position count, so a corpse + * never blocks the line. A slow-polling live waiter never reads as dead + * (ADR-0008 § Position-scaled admission polls). */ -export const ADMISSION_WAITER_TTL_MS = 100_000; +export const ADMISSION_WAITER_TTL_MS = 3 * ADMISSION_POLL_MAX_MS; /** * One D1 statement, retried twice on a transient error then surfaced — the @@ -158,8 +194,10 @@ export interface RunAdmissionStore { /** * One claim attempt: try the atomic FIFO claim; when it does not land, - * refresh our waiter heartbeat and read our queue position + the pool's - * busy count. Returns the observation (+ the clock it was taken at) for + * refresh our waiter heartbeat and read our queue position, the pool's + * busy count, and whether the burn ceiling (not the cap) held us back. A + * landed claim also stamps `executions.admitted_at` — the start of the + * container-active window the meter prices. Returns the observation (+ the clock it was taken at) for * the pure `decideAdmission` — the caller runs the whole attempt inside * one durable `step.do`, so everything here is checkpointed together. */ @@ -180,18 +218,33 @@ export interface RunAdmissionStore { readonly release: (executionId: string) => Effect.Effect; } +/** + * The rank of a class in the FIFO line: interactive (0) ahead of batch (1). A + * row with no class (pre-0011) ranks as batch. The SQL computes the same rank + * from `run_admissions.admission_class` with {@link CLASS_RANK_SQL}. + */ +const classRank = (admissionClass: AdmissionClass): number => (admissionClass === "interactive" ? 0 : 1); +const CLASS_RANK_SQL = (alias: string): string => + `(CASE WHEN ${alias}admission_class = 'interactive' THEN 0 ELSE 1 END)`; + /** * Build a D1-backed run admission store. `now` is injectable so tests can * drive the clock; production passes `Date.now` (always read inside the * caller's `step.do`, so replays see the checkpointed value). `cap` is the * per-pool slot ceiling — the dispatcher resolves it from the - * `ADMISSION_CAP` wrangler var via {@link resolveAdmissionCap}. + * `ADMISSION_CAP` wrangler var via {@link resolveAdmissionCap}. `burn` is + * this claimant's hourly rate and the deploy's burn ceiling; the default + * {@link BURN_UNLIMITED} admits on the cap alone. `admissionClass` is the + * claimant's line: interactive waiters are admitted ahead of batch ones. */ export const makeRunAdmissionD1 = ( db: D1Database, now: () => number = Date.now, cap: number = ADMISSION_CAP_DEFAULT, + burn: BurnCeiling = BURN_UNLIMITED, + admissionClass: AdmissionClass = ADMISSION_CLASS_DEFAULT, ): RunAdmissionStore => { + const rank = classRank(admissionClass); const insertQueued = ( executionId: string, pool: AdmissionPool, @@ -206,15 +259,24 @@ export const makeRunAdmissionD1 = ( // GC'd waiter row restored mid-loop keeps its original FIFO key) → // the parent's row (matrix-children priority inheritance) → now. `INSERT OR IGNORE INTO run_admissions - (execution_id, pool, state, enqueued_at, heartbeat_at) + (execution_id, pool, state, enqueued_at, heartbeat_at, hourly_rate_micro_usd, + admission_class) VALUES (?1, ?2, 'queued', COALESCE(?3, (SELECT enqueued_at FROM run_admissions WHERE execution_id = ?4), ?5), - ?5)`, + ?5, ?6, ?7)`, + ) + .bind( + executionId, + pool, + enqueuedAt ?? null, + parentExecutionId ?? null, + ts, + burn.hourlyRateMicroUsd, + admissionClass, ) - .bind(executionId, pool, enqueuedAt ?? null, parentExecutionId ?? null, ts) .run(), ).pipe(Effect.asVoid); @@ -245,9 +307,18 @@ export const makeRunAdmissionD1 = ( * are below the cap (stale heartbeats stop counting — a crashed run's * slot frees itself after the TTL); * - the NOT EXISTS is the no-barge FIFO check among LIVE waiters: nobody - * with an earlier `enqueued_at` (execution-id ULID as the tiebreak) may - * be skipped, but a waiter silent past `ADMISSION_WAITER_TTL_MS` no - * longer blocks the line. + * ahead of us may be skipped. "Ahead" is a live waiter of a higher + * class (interactive ahead of batch), or of our class with an earlier + * `enqueued_at` (execution-id ULID as the tiebreak). A waiter silent + * past `ADMISSION_WAITER_TTL_MS` no longer blocks the line; + * - the burn-ceiling clause admits only while the hourly rates of LIVE + * admitted rows in EVERY pool, plus ours, fit under the ceiling. An + * empty admitted set admits anyone, so a claimant whose own rate + * exceeds the ceiling runs alone rather than waiting forever; a NULL + * ceiling disables the clause. `fitsBurnCeiling` (core spend.ts) is its + * pure mirror. The ceiling never reorders the line: a batch claimant + * whose lower rate would fit still waits behind a live interactive + * waiter the ceiling holds. * * `meta.changes === 1` ⇒ our row flipped to 'admitted' — we hold a slot. */ @@ -267,8 +338,15 @@ export const makeRunAdmissionD1 = ( AND NOT EXISTS (SELECT 1 FROM run_admissions o WHERE o.pool = ?3 AND o.state = 'queued' AND o.heartbeat_at > ?6 - AND (o.enqueued_at < ?7 - OR (o.enqueued_at = ?7 AND o.execution_id < ?1)))`, + AND (${CLASS_RANK_SQL("o.")} < ?10 + OR (${CLASS_RANK_SQL("o.")} = ?10 + AND (o.enqueued_at < ?7 + OR (o.enqueued_at = ?7 AND o.execution_id < ?1))))) + AND (?8 IS NULL + OR NOT EXISTS (SELECT 1 FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?4) + OR (SELECT COALESCE(SUM(hourly_rate_micro_usd), 0) FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?4) + ?9 <= ?8)`, ) .bind( executionId, @@ -278,10 +356,28 @@ export const makeRunAdmissionD1 = ( cap, ts - ADMISSION_WAITER_TTL_MS, enqueuedAt, + burn.burnMaxMicroUsdPerHour, + burn.hourlyRateMicroUsd, + rank, ) .run(), ).pipe(Effect.map((result) => result.meta.changes === 1)); + // Stamp the start of the container-active window on the execution row from + // our admitted row. COALESCE keeps the first stamp across a replay. + const markAdmitted = (executionId: string): Effect.Effect => + d1(() => + db + .prepare( + `UPDATE executions + SET admitted_at = COALESCE(admitted_at, + (SELECT admitted_at FROM run_admissions WHERE execution_id = ?1)) + WHERE id = ?1`, + ) + .bind(executionId) + .run(), + ).pipe(Effect.asVoid); + const attempt = ( executionId: string, pool: AdmissionPool, @@ -302,6 +398,7 @@ export const makeRunAdmissionD1 = ( ); if (own?.state === "admitted") { yield* heartbeat(executionId); + yield* markAdmitted(executionId); return { admitted: true, position: 0, poolBusy: 0, now: ts }; } // Self-heal: a waiter paused past the admitted TTL has had its row @@ -311,11 +408,16 @@ export const makeRunAdmissionD1 = ( yield* insertQueued(executionId, pool, enqueuedAt, undefined, ts); } const claimed = yield* tryClaim(executionId, pool, enqueuedAt, ts); - if (claimed) return { admitted: true, position: 0, poolBusy: 0, now: ts }; + if (claimed) { + yield* markAdmitted(executionId); + return { admitted: true, position: 0, poolBusy: 0, now: ts }; + } // Not admitted: refresh our waiter heartbeat (so we keep blocking // bargers and counting in peers' position reads), then observe where - // we stand — live queued rows ahead of us + live admitted rows. + // we stand — live queued rows ahead of us (higher class, or our class + // and earlier), live admitted rows in our pool, and the live admitted + // burn across every pool. yield* heartbeat(executionId); const counts = yield* d1(() => db @@ -324,19 +426,37 @@ export const makeRunAdmissionD1 = ( (SELECT COUNT(*) FROM run_admissions WHERE pool = ?1 AND state = 'queued' AND heartbeat_at > ?2 - AND (enqueued_at < ?3 - OR (enqueued_at = ?3 AND execution_id < ?4))) AS position, + AND (${CLASS_RANK_SQL("")} < ?6 + OR (${CLASS_RANK_SQL("")} = ?6 + AND (enqueued_at < ?3 + OR (enqueued_at = ?3 AND execution_id < ?4))))) AS position, (SELECT COUNT(*) FROM run_admissions WHERE pool = ?1 AND state = 'admitted' - AND heartbeat_at > ?5) AS busy`, + AND heartbeat_at > ?5) AS busy, + (SELECT COUNT(*) FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?5) AS admitted_count, + (SELECT COALESCE(SUM(hourly_rate_micro_usd), 0) FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?5) AS admitted_rate`, ) - .bind(pool, ts - ADMISSION_WAITER_TTL_MS, enqueuedAt, executionId, ts - ADMISSION_TTL_MS) - .first<{ position: number; busy: number }>(), + .bind(pool, ts - ADMISSION_WAITER_TTL_MS, enqueuedAt, executionId, ts - ADMISSION_TTL_MS, rank) + .first<{ position: number; busy: number; admitted_count: number; admitted_rate: number }>(), ); + const poolBusy = counts?.busy ?? 0; + // The ceiling held us when our pool had a free slot but the burn did not + // fit. A full pool is the cap's hold, whatever the burn. + const ceilingHeld = + poolBusy < cap && + !fitsBurnCeiling({ + admittedMicroUsdPerHour: counts?.admitted_rate ?? 0, + admittedCount: counts?.admitted_count ?? 0, + ownMicroUsdPerHour: burn.hourlyRateMicroUsd, + burnMaxMicroUsdPerHour: burn.burnMaxMicroUsdPerHour, + }); return { admitted: false, position: counts?.position ?? 0, - poolBusy: counts?.busy ?? 0, + poolBusy, + ...(ceilingHeld ? { ceilingHeld } : {}), now: ts, }; }); diff --git a/packages/runtime-cf/src/run-admission-d1.workers.test.ts b/packages/runtime-cf/src/run-admission-d1.workers.test.ts index ec571827..b4901c32 100644 --- a/packages/runtime-cf/src/run-admission-d1.workers.test.ts +++ b/packages/runtime-cf/src/run-admission-d1.workers.test.ts @@ -16,7 +16,13 @@ import { Effect } from "effect"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { decideAdmission, type AdmissionPool } from "@fractalboxdev/flare-dispatch-core"; +import { + ADMISSION_POLL_MAX_MS, + containerMicroUsdPerHour, + decideAdmission, + type AdmissionClass, + type AdmissionPool, +} from "@fractalboxdev/flare-dispatch-core"; import { ADMISSION_CAP_DEFAULT, ADMISSION_MAX_QUEUE_AGE_MS, @@ -27,6 +33,7 @@ import { resolvePoolAdmissionCap, RELEASE_ADMISSION_CAP_DEFAULT, AGENT_ADMISSION_CAP_DEFAULT, + BROWSER_ADMISSION_CAP_DEFAULT, } from "./run-admission-d1"; import { makeTestBindings, type TestBindings } from "./test-support-workers"; @@ -346,10 +353,299 @@ describe("resolvePoolAdmissionCap — one cap per pool", () => { expect(resolvePoolAdmissionCap("agent", { ADMISSION_CAP: "4", ADMISSION_CAP_AGENT: "12" })).toBe(4); }); - it("leaves the lean and browser pools on the shared cap", () => { - for (const pool of ["lean", "browser"] as const) { - expect(resolvePoolAdmissionCap(pool, { ADMISSION_CAP: "16", ADMISSION_CAP_AGENT: "8", ADMISSION_CAP_RELEASE: "8" })).toBe(16); - expect(resolvePoolAdmissionCap(pool, {})).toBe(ADMISSION_CAP_DEFAULT); + it("caps the browser pool by ADMISSION_CAP_BROWSER under the shared cap", () => { + expect(BROWSER_ADMISSION_CAP_DEFAULT).toBe(16); + expect(resolvePoolAdmissionCap("browser", { ADMISSION_CAP: "16" })).toBe(16); + expect(resolvePoolAdmissionCap("browser", { ADMISSION_CAP: "16", ADMISSION_CAP_BROWSER: "6" })).toBe(6); + expect(resolvePoolAdmissionCap("browser", { ADMISSION_CAP: "4", ADMISSION_CAP_BROWSER: "12" })).toBe(4); + expect(resolvePoolAdmissionCap("browser", { ADMISSION_CAP: "16", ADMISSION_CAP_BROWSER: "six" })).toBe(16); + }); + + it("leaves the lean pool on the shared cap", () => { + expect( + resolvePoolAdmissionCap("lean", { + ADMISSION_CAP: "16", + ADMISSION_CAP_AGENT: "8", + ADMISSION_CAP_BROWSER: "6", + ADMISSION_CAP_RELEASE: "8", + }), + ).toBe(16); + expect(resolvePoolAdmissionCap("lean", {})).toBe(ADMISSION_CAP_DEFAULT); + }); + + it("keeps a waiter live across three of the longest poll sleeps", () => { + expect(ADMISSION_WAITER_TTL_MS).toBeGreaterThanOrEqual(3 * ADMISSION_POLL_MAX_MS); + }); +}); + +// The burn ceiling inside the claim (specs/adr/0006-spend-guard.md § Burn +// ceiling): admitted hourly rates across EVERY pool plus the claimant's own +// must fit, an oversized claimant admits only into an empty set, and a +// non-landing claim reports whether the ceiling or the cap held it. +describe("makeRunAdmissionD1 — the burn ceiling", () => { + let bindings: TestBindings; + beforeEach(async () => { + bindings = await makeTestBindings(); + }); + afterEach(async () => { + await bindings.dispose(); + }); + + const S4 = containerMicroUsdPerHour("standard-4"); // 401_040 µ$/h + const S2 = containerMicroUsdPerHour("standard-2"); // 129_024 µ$/h + const CEILING = 1_500_000; + + const seedAdmitted = (executionId: string, pool: AdmissionPool, rate: number) => + bindings.db + .prepare( + `INSERT INTO run_admissions + (execution_id, pool, state, enqueued_at, admitted_at, heartbeat_at, hourly_rate_micro_usd) + VALUES (?, ?, 'admitted', ?, ?, ?, ?)`, + ) + .bind(executionId, pool, T0 - 1_000, T0 - 1_000, T0, rate) + .run(); + + const claim = async (executionId: string, pool: AdmissionPool, rate: number, ceiling: number | null = CEILING) => { + const store = makeRunAdmissionD1(bindings.db, () => T0, ADMISSION_CAP_DEFAULT, { + hourlyRateMicroUsd: rate, + burnMaxMicroUsdPerHour: ceiling, + }); + const { enqueuedAt } = await Effect.runPromise(store.enqueue(executionId, pool)); + return Effect.runPromise(store.attempt(executionId, pool, enqueuedAt)); + }; + + it("records the claimant's hourly rate on its row", async () => { + await claim(A, "lean", S4); + const row = await bindings.db + .prepare(`SELECT hourly_rate_micro_usd FROM run_admissions WHERE execution_id = ?`) + .bind(A) + .first<{ hourly_rate_micro_usd: number }>(); + expect(row?.hourly_rate_micro_usd).toBe(S4); + }); + + it("admits while the admitted burn plus its own fits, across pools", async () => { + await seedAdmitted("peer-lean", "lean", S4); + await seedAdmitted("peer-release", "release", S4); + expect((await claim(A, "browser", S4)).admitted).toBe(true); + }); + + it("holds a claimant the ceiling cannot fit, with a free slot in its pool", async () => { + // Three standard-4 rows admitted in other pools: 1.203 $/h + 0.401 > 1.50. + await seedAdmitted("peer-1", "release", S4); + await seedAdmitted("peer-2", "release", S4); + await seedAdmitted("peer-3", "browser", S4); + const observed = await claim(A, "lean", S4); + expect(observed).toMatchObject({ admitted: false, poolBusy: 0, ceilingHeld: true }); + // A cheaper claimant still fits the headroom. + expect((await claim(B, "browser", S2)).admitted).toBe(true); + }); + + it("admits a claimant whose own rate exceeds the ceiling only into an empty admitted set", async () => { + expect((await claim(A, "lean", 2_000_000)).admitted).toBe(true); + const observed = await claim(B, "lean", 2_000_000); + expect(observed).toMatchObject({ admitted: false, ceilingHeld: true }); + }); + + it("stops counting a stale admitted row's burn", async () => { + await bindings.db + .prepare( + `INSERT INTO run_admissions + (execution_id, pool, state, enqueued_at, admitted_at, heartbeat_at, hourly_rate_micro_usd) + VALUES ('corpse', 'lean', 'admitted', ?, ?, ?, ?)`, + ) + .bind(T0 - 1_000, T0 - 1_000, T0 - ADMISSION_TTL_MS - 1, CEILING) + .run(); + expect((await claim(A, "lean", S4)).admitted).toBe(true); + }); + + it("admits on the cap alone with no ceiling", async () => { + await seedAdmitted("peer-1", "release", 10_000_000); + expect((await claim(A, "lean", S4, null)).admitted).toBe(true); + }); + + it("reports the cap, not the ceiling, when the pool is full", async () => { + for (let i = 0; i < ADMISSION_CAP_DEFAULT; i++) { + await seedAdmitted(`peer-${String(i).padStart(2, "0")}`, "lean", S4); } + const observed = await claim(A, "lean", S4); + expect(observed.admitted).toBe(false); + expect(observed.ceilingHeld).toBeUndefined(); + }); + + it("stamps executions.admitted_at from the claim, once", async () => { + await bindings.db + .prepare( + `INSERT INTO executions (id, run, repo, ref, sha, status, started_at, input_json) + VALUES (?, 'check', 'owner/name', 'refs/heads/main', 'abc', 'running', ?, '{}')`, + ) + .bind(A, T0 - 5_000) + .run(); + let clock = T0; + const store = makeRunAdmissionD1(bindings.db, () => clock); + const { enqueuedAt } = await Effect.runPromise(store.enqueue(A, "lean")); + await Effect.runPromise(store.attempt(A, "lean", enqueuedAt)); + // A replayed attempt later keeps the first stamp. + clock = T0 + 60_000; + await Effect.runPromise(store.attempt(A, "lean", enqueuedAt)); + const row = await bindings.db + .prepare(`SELECT admitted_at FROM executions WHERE id = ?`) + .bind(A) + .first<{ admitted_at: number }>(); + expect(row?.admitted_at).toBe(T0); + }); +}); + +// The interactive admission class (specs/adr/0009-fast-trustworthy-verdicts.md +// § Interactive admission class): a claim admits the oldest live interactive +// waiter before any batch waiter; FIFO and the queue position apply within a +// class, interactive ahead of batch; the burn ceiling never reorders the line. +describe("makeRunAdmissionD1 — the interactive admission class", () => { + let bindings: TestBindings; + beforeEach(async () => { + bindings = await makeTestBindings(); + }); + afterEach(async () => { + await bindings.dispose(); + }); + + const BATCH_OLD = "01ARZ3NDEKTSV4RRFFQ69G5F01"; + const BATCH_NEW = "01ARZ3NDEKTSV4RRFFQ69G5F02"; + const INTER_1 = "01ARZ3NDEKTSV4RRFFQ69G5F03"; + const INTER_2 = "01ARZ3NDEKTSV4RRFFQ69G5F04"; + + const storeFor = ( + admissionClass: AdmissionClass, + clock: () => number, + cap = 1, + burn = { hourlyRateMicroUsd: 0, burnMaxMicroUsdPerHour: null as number | null }, + ) => makeRunAdmissionD1(bindings.db, clock, cap, burn, admissionClass); + + /** Enqueue `id` at `at` in `admissionClass` and return its store + FIFO key. */ + const enqueueAt = async ( + id: string, + admissionClass: AdmissionClass, + at: number, + cap = 1, + burn?: { hourlyRateMicroUsd: number; burnMaxMicroUsdPerHour: number | null }, + ) => { + const store = storeFor(admissionClass, () => at, cap, burn); + const { enqueuedAt } = await Effect.runPromise(store.enqueue(id, "lean")); + return { store, enqueuedAt }; + }; + + const attemptAt = ( + id: string, + admissionClass: AdmissionClass, + enqueuedAt: number, + at: number, + cap = 1, + burn?: { hourlyRateMicroUsd: number; burnMaxMicroUsdPerHour: number | null }, + ) => Effect.runPromise(storeFor(admissionClass, () => at, cap, burn).attempt(id, "lean", enqueuedAt)); + + const holdSlot = () => + bindings.db + .prepare( + `INSERT INTO run_admissions (execution_id, pool, state, enqueued_at, admitted_at, heartbeat_at) + VALUES ('holder', 'lean', 'admitted', ?, ?, ?)`, + ) + .bind(T0 - 5_000, T0 - 5_000, T0) + .run(); + const freeSlot = () => bindings.db.prepare(`DELETE FROM run_admissions WHERE execution_id = 'holder'`).run(); + + it("records the class on the row; a store without one enqueues batch", async () => { + await enqueueAt(INTER_1, "interactive", T0); + await Effect.runPromise(makeRunAdmissionD1(bindings.db, () => T0).enqueue(BATCH_OLD, "lean")); + const classes = await bindings.db + .prepare(`SELECT execution_id, admission_class FROM run_admissions ORDER BY execution_id`) + .all<{ execution_id: string; admission_class: string }>(); + expect(classes.results).toEqual([ + { execution_id: BATCH_OLD, admission_class: "batch" }, + { execution_id: INTER_1, admission_class: "interactive" }, + ]); + }); + + it("admits a later interactive waiter before an earlier batch waiter", async () => { + await holdSlot(); + const batch = await enqueueAt(BATCH_OLD, "batch", T0 - 60_000); + const inter = await enqueueAt(INTER_1, "interactive", T0); + // Full pool: the interactive waiter is first in line, the batch one second. + expect(await attemptAt(INTER_1, "interactive", inter.enqueuedAt, T0 + 1)).toMatchObject({ + admitted: false, + position: 0, + }); + expect(await attemptAt(BATCH_OLD, "batch", batch.enqueuedAt, T0 + 2)).toMatchObject({ + admitted: false, + position: 1, + }); + await freeSlot(); + // The batch waiter cannot barge past the live interactive one. + expect((await attemptAt(BATCH_OLD, "batch", batch.enqueuedAt, T0 + 3)).admitted).toBe(false); + expect((await attemptAt(INTER_1, "interactive", inter.enqueuedAt, T0 + 4)).admitted).toBe(true); + }); + + it("keeps FIFO within the interactive class", async () => { + await holdSlot(); + const first = await enqueueAt(INTER_1, "interactive", T0 - 10_000); + const second = await enqueueAt(INTER_2, "interactive", T0); + expect((await attemptAt(INTER_2, "interactive", second.enqueuedAt, T0 + 1)).position).toBe(1); + await freeSlot(); + expect((await attemptAt(INTER_2, "interactive", second.enqueuedAt, T0 + 2)).admitted).toBe(false); + expect((await attemptAt(INTER_1, "interactive", first.enqueuedAt, T0 + 3)).admitted).toBe(true); + }); + + it("keeps FIFO within the batch class, behind every interactive waiter", async () => { + await holdSlot(); + await enqueueAt(INTER_1, "interactive", T0); + await enqueueAt(BATCH_OLD, "batch", T0 - 20_000); + const newer = await enqueueAt(BATCH_NEW, "batch", T0 - 10_000); + expect((await attemptAt(BATCH_NEW, "batch", newer.enqueuedAt, T0 + 1)).position).toBe(2); + }); + + it("ranks a pre-0011 row with no class as batch", async () => { + await holdSlot(); + await bindings.db + .prepare( + `INSERT INTO run_admissions (execution_id, pool, state, enqueued_at, heartbeat_at) + VALUES (?, 'lean', 'queued', ?, ?)`, + ) + .bind(BATCH_OLD, T0 - 60_000, T0) + .run(); + const inter = await enqueueAt(INTER_1, "interactive", T0); + expect((await attemptAt(INTER_1, "interactive", inter.enqueuedAt, T0 + 1)).position).toBe(0); + const batch = await enqueueAt(BATCH_NEW, "batch", T0); + expect((await attemptAt(BATCH_NEW, "batch", batch.enqueuedAt, T0 + 2)).position).toBe(2); + }); + + it("stops a dead interactive waiter from blocking the batch line", async () => { + await enqueueAt(INTER_1, "interactive", T0 - ADMISSION_WAITER_TTL_MS - 10); + const batch = await enqueueAt(BATCH_OLD, "batch", T0); + expect((await attemptAt(BATCH_OLD, "batch", batch.enqueuedAt, T0 + 1)).admitted).toBe(true); + }); + + it("holds a batch claimant the ceiling would fit behind an interactive waiter it holds", async () => { + // One admitted row at 1.0 $/h; ceiling 1.5 $/h. The interactive waiter at + // 0.8 $/h does not fit; the batch waiter at 0.2 $/h would, but stays in line. + const ceiling = 1_500_000; + await bindings.db + .prepare( + `INSERT INTO run_admissions + (execution_id, pool, state, enqueued_at, admitted_at, heartbeat_at, hourly_rate_micro_usd) + VALUES ('holder', 'release', 'admitted', ?, ?, ?, 1000000)`, + ) + .bind(T0 - 5_000, T0 - 5_000, T0) + .run(); + const interBurn = { hourlyRateMicroUsd: 800_000, burnMaxMicroUsdPerHour: ceiling }; + const batchBurn = { hourlyRateMicroUsd: 200_000, burnMaxMicroUsdPerHour: ceiling }; + const inter = await enqueueAt(INTER_1, "interactive", T0, 16, interBurn); + const batch = await enqueueAt(BATCH_OLD, "batch", T0 - 60_000, 16, batchBurn); + expect(await attemptAt(INTER_1, "interactive", inter.enqueuedAt, T0 + 1, 16, interBurn)).toMatchObject({ + admitted: false, + ceilingHeld: true, + }); + expect((await attemptAt(BATCH_OLD, "batch", batch.enqueuedAt, T0 + 2, 16, batchBurn)).admitted).toBe(false); + // Burn frees: the interactive waiter admits first, then the batch one fits. + await freeSlot(); + expect((await attemptAt(INTER_1, "interactive", inter.enqueuedAt, T0 + 3, 16, interBurn)).admitted).toBe(true); + expect((await attemptAt(BATCH_OLD, "batch", batch.enqueuedAt, T0 + 4, 16, batchBurn)).admitted).toBe(true); }); }); diff --git a/packages/runtime-cf/src/runtime.ts b/packages/runtime-cf/src/runtime.ts index 53e18673..766711d5 100644 --- a/packages/runtime-cf/src/runtime.ts +++ b/packages/runtime-cf/src/runtime.ts @@ -97,6 +97,11 @@ export type CFRuntimeLiveOptions = { * required Dispatcher binding, so the production call site always passes it. */ readonly runsWorkflow?: WorkflowBindingLike; + /** + * The monthly-budget read a fan-out runs before it spawns children + * (ADR-0006) — see `ChildRunsLiveConfig.spendGate`. Absent = no spend guard. + */ + readonly spendGate?: () => Promise; /** This execution's ULID — namespaces D1 rows, R2 keys, the sandbox id. */ readonly executionId: string; /** repo/ref/sha/input the `executions` row requires. */ @@ -413,6 +418,7 @@ export const makeCFRuntimeLive = ( workflow: opts.runsWorkflow, db: opts.db, parentExecutionId: opts.executionId, + ...(opts.spendGate !== undefined ? { spendGate: opts.spendGate } : {}), // Children inherit the parent's public origin so their artifact // links are absolute too (a child posts its own check-run). ...(opts.publicOrigin !== undefined ? { origin: opts.publicOrigin } : {}), diff --git a/packages/runtime-cf/src/spend-d1.ts b/packages/runtime-cf/src/spend-d1.ts new file mode 100644 index 00000000..c551d7c3 --- /dev/null +++ b/packages/runtime-cf/src/spend-d1.ts @@ -0,0 +1,179 @@ +// @fractalboxdev/flare-dispatch-runtime-cf — spend guard I/O: the month-to-date read and the limits. +// +// The I/O shell around the pure spend math (`@fractalboxdev/flare-dispatch-core` +// spend.ts): one indexed D1 query for month-to-date spend and the admitted +// burn, the `CONFIG_KV` overrides for the limits, and the once-per-month +// warning marker. The dispatcher calls these inside durable `step.do` bodies, +// so a replay reads the checkpointed answer rather than a fresh one. +// +// Spec: specs/adr/0006-spend-guard.md § Spend read, § Monthly budget, +// § Warning and visibility. + +import { Effect, Schedule } from "effect"; +import { + type BudgetDecision, + decideBudget, + formatUsd, + monthToDateMicroUsd, + resolveSpendLimits, + SPEND_BUDGET_KV_KEY, + SPEND_BURN_KV_KEY, + type SpendLimits, + spendMonth, + type SpendMonth, + type SpendReading, + type SpendVars, + spendWarnedKvKey, +} from "@fractalboxdev/flare-dispatch-core"; +import { ADMISSION_TTL_MS } from "./run-admission-d1"; + +const MS_PER_HOUR = 3_600_000; + +/** + * Read month-to-date spend and the live admitted burn — one statement: + * + * * `finished` — `cost_micro_usd` of executions admitted since the month + * started (the `executions_admitted_at` range scan); + * * `in_flight_rate_ms` — Σ hourly rate × ms since admission over LIVE + * admitted `run_admissions` rows admitted this month. A row is deleted at + * release, before the finish path writes `cost_micro_usd`, so no execution + * counts in both terms; + * * `admitted_rate` / `admitted_count` — the live admitted set the burn + * ceiling sums, across every pool. + * + * A row written before migration 0011 has a NULL rate and counts as zero. + */ +export const readSpend = (db: D1Database, nowMs: number): Effect.Effect => + Effect.tryPromise({ + try: () => + db + .prepare( + `SELECT + (SELECT COALESCE(SUM(cost_micro_usd), 0) FROM executions + WHERE admitted_at >= ?1 AND cost_micro_usd IS NOT NULL) AS finished, + (SELECT COALESCE(SUM(COALESCE(hourly_rate_micro_usd, 0) * (?2 - admitted_at)), 0) + FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?3 + AND admitted_at >= ?1 AND admitted_at <= ?2) AS in_flight_rate_ms, + (SELECT COALESCE(SUM(hourly_rate_micro_usd), 0) FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?3) AS admitted_rate, + (SELECT COUNT(*) FROM run_admissions + WHERE state = 'admitted' AND heartbeat_at > ?3) AS admitted_count`, + ) + .bind(spendMonth(nowMs).startMs, nowMs, nowMs - ADMISSION_TTL_MS) + .first<{ + finished: number; + in_flight_rate_ms: number; + admitted_rate: number; + admitted_count: number; + }>(), + catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + }).pipe( + Effect.retry(Schedule.recurs(2).pipe(Schedule.addDelay(() => "200 millis"))), + Effect.map((row) => ({ + finishedMicroUsd: row?.finished ?? 0, + inFlightMicroUsd: Math.round((row?.in_flight_rate_ms ?? 0) / MS_PER_HOUR), + admittedMicroUsdPerHour: row?.admitted_rate ?? 0, + admittedCount: row?.admitted_count ?? 0, + })), + ); + +/** + * Resolve the limits: `CONFIG_KV` `spend.budgetMonthlyUsd` / + * `spend.burnMaxUsdPerHour` override the vars, and the vars override the code + * defaults. A KV read that fails falls back to the vars and logs: the limit + * then binds at the configured value, never at none. + */ +export const readSpendLimits = ( + kv: KVNamespace | undefined, + vars: SpendVars, +): Effect.Effect => + (kv === undefined + ? Effect.succeed({}) + : Effect.tryPromise(async () => { + const [budgetMonthlyUsd, burnMaxUsdPerHour] = await Promise.all([ + kv.get(SPEND_BUDGET_KV_KEY), + kv.get(SPEND_BURN_KV_KEY), + ]); + return { budgetMonthlyUsd, burnMaxUsdPerHour }; + }).pipe( + Effect.catchAll((cause) => + Effect.logWarning(`spend: CONFIG_KV limit overrides unreadable, using vars — ${cause}`).pipe( + Effect.as({}), + ), + ), + ) + ).pipe(Effect.map((overrides) => resolveSpendLimits(vars, overrides))); + +/** One spend read with its verdict against the budget. */ +export type SpendStatus = { + readonly month: SpendMonth; + readonly limits: SpendLimits; + readonly reading: SpendReading; + /** Finished + in-flight, times the overhead factor. */ + readonly monthToDateMicroUsd: number; + readonly decision: BudgetDecision; +}; + +/** Read spend and decide the monthly budget against `limits`. */ +export const readSpendStatus = ( + db: D1Database, + limits: SpendLimits, + nowMs: number, +): Effect.Effect => + Effect.map(readSpend(db, nowMs), (reading) => { + const spend = monthToDateMicroUsd(reading, limits.overheadFactor); + return { + month: spendMonth(nowMs), + limits, + reading, + monthToDateMicroUsd: spend, + decision: decideBudget(spend, limits), + }; + }); + +/** + * Send the month's one spend warning when `status` crossed the warn ratio and + * the month carries no `spend.warned:` marker yet. Writes the marker + * before notifying, so a failed notification is not retried by every later + * admission. Returns whether this call sent it. Best-effort: a KV or notify + * failure logs and returns `false`; it never touches the run. + * + * Without `CONFIG_KV` there is nowhere to mark the month, so it logs + * `spend.warn` and sends nothing. + */ +export const warnSpendOnce = (opts: { + readonly kv: KVNamespace | undefined; + readonly status: SpendStatus; + readonly notify: (text: string, monthKey: string) => Promise; +}): Effect.Effect => { + const { status } = opts; + if (status.decision._kind !== "open" || !status.decision.warn) return Effect.succeed(false); + const budget = status.limits.budgetMicroUsd ?? 0; + const text = + `FlareDispatch spend reached ${formatUsd(status.monthToDateMicroUsd)} of the ` + + `${formatUsd(budget)} monthly budget for ${status.month.key}. ` + + `Checks skip once it reaches the budget; raise it with CONFIG_KV ${SPEND_BUDGET_KV_KEY}.`; + const kv = opts.kv; + if (kv === undefined) { + return Effect.logWarning(`spend.warn ${text} (no CONFIG_KV — not notified)`).pipe(Effect.as(false)); + } + const marker = spendWarnedKvKey(status.month.key); + return Effect.gen(function* () { + const warned = yield* Effect.tryPromise(() => kv.get(marker)); + if (warned !== null) return false; + yield* Effect.tryPromise(() => + kv.put(marker, String(status.monthToDateMicroUsd), { + // Outlive the month, then expire — a stale marker never silences a later year. + expiration: Math.floor(status.month.resetsAtMs / 1000) + 7 * 24 * 3600, + }), + ); + yield* Effect.logWarning(`spend.warn ${text}`); + yield* Effect.tryPromise(() => opts.notify(text, status.month.key)); + return true; + }).pipe( + Effect.catchAll((cause) => + Effect.logWarning(`spend.warn not delivered — ${cause}`).pipe(Effect.as(false)), + ), + ); +}; diff --git a/packages/runtime-cf/src/spend-d1.workers.test.ts b/packages/runtime-cf/src/spend-d1.workers.test.ts new file mode 100644 index 00000000..3a92b7e3 --- /dev/null +++ b/packages/runtime-cf/src/spend-d1.workers.test.ts @@ -0,0 +1,118 @@ +// Integration tests for the spend guard's I/O against real D1 and KV: the +// month-to-date read, the CONFIG_KV limit overrides, and the once-per-month +// warning marker (specs/adr/0006-spend-guard.md). The pure math is covered by +// core spend.test.ts. + +import { Effect } from "effect"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { resolveSpendLimits, type SpendLimits } from "@fractalboxdev/flare-dispatch-core"; +import { readSpend, readSpendLimits, readSpendStatus, warnSpendOnce } from "./spend-d1"; +import { ADMISSION_TTL_MS } from "./run-admission-d1"; +import { makeTestBindings, type TestBindings } from "./test-support-workers"; + +const NOW = Date.UTC(2026, 9, 22, 12, 0); +const MONTH_START = Date.UTC(2026, 9, 1); +const HOUR = 3_600_000; + +describe("spend-d1", () => { + let bindings: TestBindings; + beforeEach(async () => { + bindings = await makeTestBindings(); + }); + afterEach(async () => { + await bindings.dispose(); + }); + + const seedFinished = (id: string, admittedAt: number | null, cost: number) => + bindings.db + .prepare( + `INSERT INTO executions + (id, run, repo, ref, sha, status, started_at, admitted_at, completed_at, input_json, cost_micro_usd) + VALUES (?, 'check', 'owner/name', 'refs/heads/main', 'abc', 'success', ?, ?, ?, '{}', ?)`, + ) + .bind(id, admittedAt ?? NOW, admittedAt, NOW, cost) + .run(); + + const seedAdmitted = (id: string, admittedAt: number, rate: number, heartbeatAt = NOW) => + bindings.db + .prepare( + `INSERT INTO run_admissions + (execution_id, pool, state, enqueued_at, admitted_at, heartbeat_at, hourly_rate_micro_usd) + VALUES (?, 'lean', 'admitted', ?, ?, ?, ?)`, + ) + .bind(id, admittedAt, admittedAt, heartbeatAt, rate) + .run(); + + it("sums finished executions admitted this month and in-flight burn since admission", async () => { + await seedFinished("f1", MONTH_START + HOUR, 3_000_000); + await seedFinished("f2", NOW - HOUR, 1_000_000); + // Admitted last month: not this month's spend. + await seedFinished("old", MONTH_START - HOUR, 50_000_000); + // Priced before the meter (no admission stamp): not counted. + await seedFinished("legacy", null, 7_000_000); + // In flight for two hours at $0.40/h; a stale row counts for nothing. + await seedAdmitted("live", NOW - 2 * HOUR, 400_000); + await seedAdmitted("corpse", NOW - 2 * HOUR, 400_000, NOW - ADMISSION_TTL_MS - 1); + + const reading = await Effect.runPromise(readSpend(bindings.db, NOW)); + expect(reading).toEqual({ + finishedMicroUsd: 4_000_000, + inFlightMicroUsd: 800_000, + admittedMicroUsdPerHour: 400_000, + admittedCount: 1, + }); + }); + + it("closes the budget once spend times the overhead reaches it", async () => { + await seedFinished("f1", NOW - HOUR, 46_000_000); + const limits = resolveSpendLimits({}); + const status = await Effect.runPromise(readSpendStatus(bindings.db, limits, NOW)); + // 46 × 1.10 = 50.60 ≥ 50. + expect(status.monthToDateMicroUsd).toBe(50_600_000); + expect(status.decision).toEqual({ + _kind: "closed", + reason: "spend budget reached: $50.60 of $50.00 this month", + }); + expect(status.month.key).toBe("2026-10"); + }); + + it("reads the CONFIG_KV overrides over the vars", async () => { + await bindings.kv.put("spend.budgetMonthlyUsd", "unlimited"); + await bindings.kv.put("spend.burnMaxUsdPerHour", "4"); + const limits = await Effect.runPromise( + readSpendLimits(bindings.kv, { SPEND_BUDGET_MONTHLY_USD: "50", SPEND_BURN_MAX_USD_PER_HOUR: "2" }), + ); + expect(limits).toMatchObject({ budgetMicroUsd: null, burnMaxMicroUsdPerHour: 4_000_000 }); + expect(await Effect.runPromise(readSpendLimits(undefined, { SPEND_BUDGET_MONTHLY_USD: "50" }))).toMatchObject( + { budgetMicroUsd: 50_000_000 }, + ); + }); + + it("warns once per month past the warn ratio, marking the month in CONFIG_KV", async () => { + await seedFinished("f1", NOW - HOUR, 40_000_000); + const limits: SpendLimits = resolveSpendLimits({}); + const status = await Effect.runPromise(readSpendStatus(bindings.db, limits, NOW)); + const sent: string[] = []; + const notify = async (text: string, month: string) => { + sent.push(`${month}: ${text}`); + }; + + expect(await Effect.runPromise(warnSpendOnce({ kv: bindings.kv, status, notify }))).toBe(true); + expect(await Effect.runPromise(warnSpendOnce({ kv: bindings.kv, status, notify }))).toBe(false); + expect(sent).toHaveLength(1); + expect(sent[0]).toContain("2026-10: FlareDispatch spend reached $44.00 of the $50.00 monthly budget"); + expect(await bindings.kv.get("spend.warned:2026-10")).toBe("44000000"); + }); + + it("sends no warning under the warn ratio", async () => { + await seedFinished("f1", NOW - HOUR, 10_000_000); + const status = await Effect.runPromise(readSpendStatus(bindings.db, resolveSpendLimits({}), NOW)); + let calls = 0; + const warned = await Effect.runPromise( + warnSpendOnce({ kv: bindings.kv, status, notify: async () => void calls++ }), + ); + expect(warned).toBe(false); + expect(calls).toBe(0); + expect(await bindings.kv.get("spend.warned:2026-10")).toBeNull(); + }); +}); diff --git a/runs/admission-class.test.ts b/runs/admission-class.test.ts new file mode 100644 index 00000000..0d80c55b --- /dev/null +++ b/runs/admission-class.test.ts @@ -0,0 +1,29 @@ +// Catalog-level guard for `limits.admissionClass` +// (specs/adr/0009-fast-trustworthy-verdicts.md § Interactive admission class). +// +// An interactive run is admitted ahead of every batch run in its pool, so the +// set is a deliberate list: the PR verdicts and the deploy someone waits on. +// The assertion derives from the whole catalog, so a new run that declares +// `interactive` fails here until the list is updated on purpose. + +import { describe, expect, it } from "vitest"; +import type { Run } from "@fractalboxdev/flare-dispatch-core"; +import * as catalog from "./index"; + +const runs = Object.values(catalog) as ReadonlyArray>; + +describe("admissionClass catalog", () => { + it("admits exactly the waited-on runs in the interactive class", () => { + const interactive = runs + .filter((r) => r.limits.admissionClass === "interactive") + .map((r) => r.name) + .sort(); + expect(interactive).toEqual(["offload-test", "oxlint", "pr-review", "worker-deploy"]); + }); + + it("leaves every other run in the default batch class", () => { + for (const r of runs) { + expect(["interactive", "batch", undefined]).toContain(r.limits.admissionClass); + } + }); +}); diff --git a/runs/contextful-gate.test.ts b/runs/contextful-gate.test.ts index a2ff3801..5310f4d6 100644 --- a/runs/contextful-gate.test.ts +++ b/runs/contextful-gate.test.ts @@ -912,6 +912,23 @@ describe("contextful-gate", () => { }).pipe(Effect.provide(layer)); }); + it.effect("spawns no part once the monthly spend budget is reached, and skips", () => { + const reason = "spend budget reached: $90.00 of $90.00 this month"; + const { layer, handles } = makeCFRuntimeTest({ + config: { "contextful-gate.window:fractalboxdev/contextful": "1" }, + sandboxProgram: three, + childRuns: { spendClosed: reason }, + }); + return Effect.gen(function* () { + const failure = yield* Effect.flip(contextfulGate.run(input)); + expect(failure).toMatchObject({ _tag: "RunSkipped", reason: `${reason} — 3 of 3 parts not run` }); + expect(handles.childRuns.spawned).toHaveLength(0); + expect(cancelledChecks(handles)).toEqual(["pins", "workspace.compile", "formal"].map((stage) => ({ + name: `flare-dispatch/check:${stage}`, conclusion: "cancelled", summary: `not run — ${reason}`, + }))); + }).pipe(Effect.provide(layer)); + }); + it.effect("keeps spawning after an ordinary red part", () => { const { layer, handles } = makeCFRuntimeTest({ config: { "contextful-gate.window:fractalboxdev/contextful": "1" }, diff --git a/runs/contextful-gate.ts b/runs/contextful-gate.ts index 3284dc57..ea2321e5 100644 --- a/runs/contextful-gate.ts +++ b/runs/contextful-gate.ts @@ -84,8 +84,16 @@ export const parseWindow = (raw: string | undefined): number => { export const partCheckName = (stage: string): string => `flare-dispatch/check:${stage}`; export const NOT_AFFECTED_SUMMARY = "not affected by this change"; -/** Why a gate stops spawning before every part has run. */ -export type StopReason = "head superseded" | "pull request closed" | "admission backlog"; +/** + * Why a gate stops spawning before every part has run. A `spend budget + * reached: …` reason is the deploy's monthly budget closing before a window + * spawns (specs/adr/0006-spend-guard.md). + */ +export type StopReason = + | "head superseded" + | "pull request closed" + | "admission backlog" + | `spend budget reached: ${string}`; /** Whether the gated pull request is still open at the gated commit. */ export type Liveness = { readonly live: true } | { readonly live: false; readonly reason: StopReason }; @@ -491,7 +499,7 @@ export const contextfulGate = defineRun({ Effect.gen(function* () { const liveness = yield* readLiveness(input); if (!liveness.live) return { stopped: liveness.reason }; - const handles = yield* fanOut({ + const fanned = yield* fanOut({ run: "check", items: batch, concurrency: batch.length, @@ -509,8 +517,15 @@ export const contextfulGate = defineRun({ : { restore: [{ key: prebuilt.key, base: prebuilt.base, paths: prebuilt.paths }] }), secrets: [], }), - }); - return { handles, spawnedAt: yield* io.now }; + }).pipe( + // The monthly budget closed: spawn nothing more, post the + // unrun parts, and end the gate skipped with that reason. + Effect.catchTag("RunSkipped", (skip) => + Effect.succeed({ stopped: skip.reason as StopReason }), + ), + ); + if ("stopped" in fanned) return fanned; + return { handles: fanned, spawnedAt: yield* io.now }; }), ); if ("stopped" in spawn) { diff --git a/runs/contextful-release.test.ts b/runs/contextful-release.test.ts index db6865a3..a3e614cd 100644 --- a/runs/contextful-release.test.ts +++ b/runs/contextful-release.test.ts @@ -310,12 +310,13 @@ describe("contextful-release", () => { const releaseFixture = ( status: (cell: Cell, poll: number, now: number, attempt: number) => Verdict, existing: readonly string[] = [], + spendClosed?: () => string | undefined, ) => { let spawned: readonly { readonly input: unknown; readonly instanceId: string }[] = []; let now = 0; const runtime = makeCFRuntimeTest({ github: { files: { [`${repo}:Cargo.toml`]: '[workspace.package]\nversion = "0.5.0"\n' } }, - childRuns: { existing, pollFn: (ids, poll) => ids.map((id) => { + childRuns: { existing, ...(spendClosed !== undefined ? { spendClosed } : {}), pollFn: (ids, poll) => ids.map((id) => { if (id.startsWith("contextful-release-formula")) { return { executionId: id, status: "success", summaryJson: JSON.stringify({ assets: FORMULA_ASSETS, notes: "## Consumer compatibility\n" }) }; @@ -425,6 +426,45 @@ describe("contextful-release", () => { }).pipe(Effect.provide(layer)); }); + const skipOf = (exit: Exit.Exit) => { + if (!Exit.isFailure(exit)) return undefined; + const failure = Option.getOrUndefined(Cause.failureOption(exit.cause)) as + { _tag?: string; reason?: string } | undefined; + expect(failure?._tag).toBe("RunSkipped"); + return failure?.reason; + }; + const budgetReason = "spend budget reached: $90.00 of $90.00 this month"; + + it.effect("a budget closed before the first spawn starts no cell and skips the release", () => { + const { layer, handles, provideClock } = releaseFixture(() => "success", [], () => budgetReason); + return Effect.gen(function* () { + const exit = yield* Effect.exit(provideClock(contextfulRelease.run({ repo, tag: "v0.5.0", sha, dryRun: true }))); + expect(skipOf(exit)).toBe(`${budgetReason} — 10 of 10 release cells not started`); + expect(handles.childRuns.spawned).toHaveLength(0); + expect(handles.childRuns.admissionHandoffs).toBe(0); + }).pipe(Effect.provide(layer)); + }); + + it.effect("a budget that closes mid-release starts no further cell, lets the running ones settle, then skips", () => { + // The first window spawns; the budget is closed for every later spawn. + let gateCalls = 0; + const { layer, handles, provideClock } = releaseFixture( + (cell, poll) => (cell.profile === "contextful-full" && cell.target === "x86_64-unknown-linux-musl") || poll >= 3 + ? "success" : "running", + [], + () => (gateCalls++ === 0 ? undefined : budgetReason), + ); + return Effect.gen(function* () { + const exit = yield* Effect.exit(provideClock(contextfulRelease.run({ repo, tag: "v0.5.0", sha, dryRun: true }))); + expect(skipOf(exit)).toBe(`${budgetReason} — 2 of 10 release cells not started`); + expect(cellsOf(handles)).toHaveLength(8); + expect(handles.childRuns.spawned.some((child) => child.run === "contextful-release-formula")).toBe(false); + // Every running cell was awaited to its terminal poll before the release ended. + const awaits = stepNames(handles).filter((name) => name.startsWith("await-cells-")); + expect(awaits.length).toBeGreaterThan(1); + }).pipe(Effect.provide(layer)); + }); + it.effect("a cell the platform lost respawns as a fresh child, at most twice, and the release completes", () => { const lost = (cell: Cell, tag: string, attempts: number) => (c: Cell, attempt: number) => c.profile === cell.profile && c.target === cell.target && attempt < attempts ? { failure: tag } : undefined; diff --git a/runs/contextful-release.ts b/runs/contextful-release.ts index ae2be4d4..e22fa093 100644 --- a/runs/contextful-release.ts +++ b/runs/contextful-release.ts @@ -1,7 +1,7 @@ import { Effect, Schema } from "effect"; import { AcceptanceFailed, artifact, checkpointSleep, defineRun, GitHubApiError, github, handoffChildAdmission, io, - isTerminalChildStatus, sandbox, step, StepFailed, type ChildStatusRecord, + isTerminalChildStatus, RunSkipped, sandbox, step, StepFailed, type ChildStatusRecord, } from "@fractalboxdev/flare-dispatch-core"; import { DURABLE_EXEC_RETRY_ON, durableExec, ensureWorkspace, fanOut, waitForAnyChildDurably, waitForChildren, workspace, @@ -436,15 +436,21 @@ export const contextfulRelease = defineRun({ const timedOut = new Set(); const cellFailed = (child: ChildStatusRecord) => child.status !== "success" || !child.summaryJson; let spawned = 0; + // The monthly budget closing at a spawn (specs/adr/0006-spend-guard.md) + // stops the window like a failed cell: no further cell starts, and the + // running ones settle before the release ends skipped. + let budgetClosed: string | undefined; for (let round = 0; queue.length > 0 || running.length > 0; round++) { - const failed = timedOut.size > 0 || [...settledByCell.values()].some(cellFailed); + const failed = budgetClosed !== undefined || timedOut.size > 0 || [...settledByCell.values()].some(cellFailed); const free = RELEASE_WINDOW - running.length; // After a failure no further cell starts; the running ones settle or // outlive their budget, so the verdict names every failed cell. They // write to the same draft release, so the parent never ends before them. if (!failed && queue.length > 0 && free > 0) { - const batch = queue.splice(0, free); - const { handles, spawnedAt } = yield* step(`spawn-cells-${spawned}`, () => Effect.gen(function* () { + const batch = queue.slice(0, free); + // The skip is caught inside the step, so it checkpoints as an answer + // rather than a failure the platform retries. + const spawn = yield* step(`spawn-cells-${spawned}`, () => Effect.gen(function* () { const handles = yield* fanOut({ run: contextfulReleaseCell.name, items: batch, @@ -453,7 +459,14 @@ export const contextfulRelease = defineRun({ releaseId: release.id, dryRun, ...(attempt > 0 ? { attempt } : {}) }), }); return { handles, spawnedAt: yield* io.now }; - })); + }).pipe(Effect.catchTag("RunSkipped", (skip) => Effect.succeed({ budgetClosed: skip.reason })))); + if ("budgetClosed" in spawn) { + budgetClosed = spawn.budgetClosed; + if (running.length === 0) break; + continue; + } + const { handles, spawnedAt } = spawn; + queue.splice(0, batch.length); if (spawned === 0) yield* step("handoff-admission", () => handoffChildAdmission()); batch.forEach(({ cell, attempt }, index) => running.push({ index: CELLS.indexOf(cell), @@ -499,6 +512,11 @@ export const contextfulRelease = defineRun({ const child = settledByCell.get(index); return child !== undefined && cellFailed(child); }); + if (budgetClosed !== undefined && failedCells.length === 0 && timedOut.size === 0) { + return yield* Effect.fail(new RunSkipped({ + reason: `${budgetClosed} — ${queue.length} of ${CELLS.length} release cells not started`, + })); + } if (failedCells.length > 0 || timedOut.size > 0) { // A typed failure, not a defect: the check-run renders `summaryMd`, so // the verdict names every failed, timed-out and unstarted cell there. @@ -507,7 +525,7 @@ export const contextfulRelease = defineRun({ const verdict = [ ["Failed", failedCells], ["Timed out", CELLS.filter((_, index) => timedOut.has(index))], - ["Not started", queue.map(({ cell }) => cell)], + [budgetClosed === undefined ? "Not started" : `Not started (${budgetClosed})`, queue.map(({ cell }) => cell)], ] as const; return yield* Effect.fail(new AcceptanceFailed({ exitCode: 1, summaryMd: `**Release cells**\n\n${verdict .filter(([, cells]) => cells.length > 0) @@ -528,11 +546,15 @@ export const contextfulRelease = defineRun({ return yield* Effect.die(new Error("release tag moved during build")); } } - const formulaHandle = yield* step("spawn-formula", () => fanOut({ run: contextfulReleaseFormula.name, + const formulaSpawn = yield* step("spawn-formula", () => fanOut({ run: contextfulReleaseFormula.name, items: [JSON.stringify(manifests)], toInput: (manifestJson) => ({ repo: input.repo, tag: input.tag, sha: tag.commitSha, releaseId: release.id, dryRun, manifests: manifestJson }), - })); + }).pipe(Effect.catchTag("RunSkipped", (skip) => Effect.succeed({ budgetClosed: skip.reason })))); + if ("budgetClosed" in formulaSpawn) { + return yield* Effect.fail(new RunSkipped({ reason: `${formulaSpawn.budgetClosed} — release formula not started` })); + } + const formulaHandle = formulaSpawn; const formulaChildren = yield* step("await-formula", () => waitForChildren({ ids: formulaHandle.map((handle) => handle.executionId), timeout: "8 hours", })); diff --git a/runs/offload-test.ts b/runs/offload-test.ts index b5077783..476dd78f 100644 --- a/runs/offload-test.ts +++ b/runs/offload-test.ts @@ -511,6 +511,8 @@ export const offloadTest = defineRun({ // dispatcher. One documented 1800 s exec plus checkout, install and // upload fits; aligned retries of that exec do not (issue #42). maxDurationSec: 3600, + // A PR verdict someone waits on: admitted ahead of batch work. + admissionClass: "interactive", }, run: (input) => diff --git a/runs/oxlint.ts b/runs/oxlint.ts index 59dcbbfc..e816d793 100644 --- a/runs/oxlint.ts +++ b/runs/oxlint.ts @@ -204,6 +204,8 @@ export const oxlint = defineRun({ // Wall-time ceiling — lint is cheap, but a cold `npx` fetch + a huge tree // gets headroom. maxDurationSec: 600, + // A PR verdict someone waits on: admitted ahead of batch work. + admissionClass: "interactive", }, run: (input) => diff --git a/runs/pr-review.ts b/runs/pr-review.ts index 68e4a4b8..22f66e6d 100644 --- a/runs/pr-review.ts +++ b/runs/pr-review.ts @@ -304,7 +304,8 @@ export const prReview = defineRun({ outputs: ReviewOutput, - limits: { maxDurationSec: 1500, maxConcurrency: FULL_AGENTS.length }, + // A PR verdict someone waits on: admitted ahead of batch work. + limits: { maxDurationSec: 1500, maxConcurrency: FULL_AGENTS.length, admissionClass: "interactive" }, // At most one review per PR per window, across BOTH dispatch paths // (webhook + Action). Default 60 min; override via CONFIG_KV diff --git a/runs/worker-deploy.ts b/runs/worker-deploy.ts index 17bb88a4..ba0bfd48 100644 --- a/runs/worker-deploy.ts +++ b/runs/worker-deploy.ts @@ -334,6 +334,8 @@ export const workerDeploy = defineRun({ limits: { maxDurationSec: 1800, + // A deploy someone waits on: admitted ahead of batch work. + admissionClass: "interactive", }, run: (input) => diff --git a/wrangler.jsonc b/wrangler.jsonc index 8bb58cd2..8911bccb 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -88,10 +88,12 @@ // mutually exclusive). If the gateway has Authenticated Gateway turned on, // also set the AI_GATEWAY_AUTH_TOKEN secret. Operator overlay. "AI_GATEWAY_ID": "your-ai-gateway", - "ADMISSION_CAP": "16", - // The agent and release pools admit at most RunSandboxAgent's and - // RunSandboxRelease's max_instances. + "ADMISSION_CAP": "48", + // The agent, browser and release pools admit at most RunSandboxAgent's, + // RunSandboxBrowser's and RunSandboxRelease's max_instances. Each cap is an + // upper bound; SPEND_BURN_MAX_USD_PER_HOUR sets the operating concurrency. "ADMISSION_CAP_AGENT": "8", + "ADMISSION_CAP_BROWSER": "16", "ADMISSION_CAP_RELEASE": "8", // Opt the human-facing viewer surfaces (incl. the GitHub App install flow) // out of the Cloudflare Access gate for the initial dogfood. Configure @@ -191,26 +193,26 @@ "image_build_context": ".", "image_vars": { "WITH_BUILD_TOOLS": "true", "WITH_RUST": "true" }, "instance_type": "standard-4", - "max_instances": 16, + "max_instances": 56, }, { "class_name": "RunSandboxBrowser", "image": "./infra/Dockerfile.sandbox", "instance_type": "standard-2", - "max_instances": 16, + "max_instances": 24, }, { "class_name": "RunSandboxAgent", "image": "./infra/Dockerfile.sandbox", "instance_type": "standard-3", - "max_instances": 8, + "max_instances": 16, }, { "class_name": "RunSandboxRelease", "image": "./infra/Dockerfile.sandbox-release", "image_build_context": ".", "instance_type": "standard-4", - "max_instances": 8, + "max_instances": 16, // Every deploy pushes a new image and rolls every container. A rollout // replaces an active instance once this grace has passed, and a // replaced release container loses its build (`SandboxProcessLost`):