From cbc92aff31c3dbeba72599bd3e2908433a721bea Mon Sep 17 00:00:00 2001 From: pelikhan Date: Thu, 5 Nov 2026 23:14:00 -0800 Subject: [PATCH 1/2] fix(work-queue): avoid redundant history validation Reuse validated canonical prefixes and native Git blob identities, reconstruct checkpoint source hashes without repeated replay, and avoid redundant compressed receipt parsing and cloning. Add bounded synthetic and captured-history benchmarks plus integrity regressions while preserving delivery deadlines and authority checks. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../work_queue_checkpoint_regression.test.cjs | 48 ++++++++++- .../js/work_queue_git_transport.test.cjs | 83 ++++++++++++++++++- .../js/work_queue_live_performance.test.cjs | 35 +++++++- actions/setup/js/work_queue_replay.cjs | 50 ++++++++--- actions/setup/js/work_queue_store.cjs | 66 +++++++++++++-- .../docs/specs/work-queue-specification.md | 18 ++++ 6 files changed, 280 insertions(+), 20 deletions(-) diff --git a/actions/setup/js/work_queue_checkpoint_regression.test.cjs b/actions/setup/js/work_queue_checkpoint_regression.test.cjs index 7367c581146..a7aae257b73 100644 --- a/actions/setup/js/work_queue_checkpoint_regression.test.cjs +++ b/actions/setup/js/work_queue_checkpoint_regression.test.cjs @@ -1,11 +1,12 @@ import { describe, it } from "vitest"; import assert from "node:assert/strict"; import { createRequire } from "node:module"; +import { deflateRawSync, inflateRawSync } from "node:zlib"; const require = createRequire(import.meta.url); const { canonical } = require("./work_queue_codec.cjs"); const { administrator, genesis, operationCommit, submission } = require("./work_queue_test_helpers.cjs"); -const { compactTransactions, prepareCheckpoint, reconstructCheckpoint, replayTransactions, serializeProjection } = require("./work_queue_replay.cjs"); +const { compactTransactions, prepareCheckpoint, reconstructCheckpoint, replayCheckpointSource, replayTransactions, serializeProjection, serializeTransactionLog } = require("./work_queue_replay.cjs"); const { legacyCheckpoint, rewriteCheckpoint } = require("./work_queue_checkpoint_regression_helpers.cjs"); const { acceptedSubmissionParameters } = require("./work_queue_dispatch.cjs"); @@ -16,6 +17,27 @@ function expiringLog() { } describe("checkpoint and expiration internal-state regressions", () => { + it("rejects noncanonical compressed receipts even when native JSON parsing produces the original snapshot", () => { + const log = [genesis()]; + log.push(submission(log, ["accepted"])); + const root = compactTransactions(log, "a".repeat(40), administrator, 100)[0]; + const text = inflateRawSync(Buffer.from(root.operations[0].state.requests_compressed, "base64")).toString("utf8"); + for (const changed of [ + text.replace(/"at":([0-9]+)/, '"at":$1,"at":$1'), + text.replace(/"at":([0-9]+)/, (_, at) => `"at":${at},"\\u0061t":${at}`), + text.replace(/"at":([0-9]+)/, '"at":$1.0'), + text.replace(/"at":([0-9]+)/, '"at":$1e0'), + text.replace('"actor":', '"\\u0061ctor":'), + ` ${text}`, + ]) { + assert.notEqual(changed, text); + assert.equal(canonical(JSON.parse(changed)), text); + const damaged = structuredClone(root); + damaged.operations[0].state.requests_compressed = deflateRawSync(Buffer.from(changed, "utf8")).toString("base64"); + assert.throws(() => replayTransactions([damaged]), /checkpoint_invalid/); + } + }); + it("recovers checkpoints written before full Policy receipt history and retains current-epoch submit recovery", () => { const log = [genesis()]; log.push(submission(log, ["accepted"])); @@ -26,6 +48,7 @@ describe("checkpoint and expiration internal-state regressions", () => { assert.notEqual(legacy[0].operations[0].state_sha256, current[0].operations[0].state_sha256); const recovered = reconstructCheckpoint(log, "a".repeat(40), administrator, 100, legacy[0].operations[0].state_sha256); assert.equal(canonical(recovered), canonical(legacy)); + assert.equal(canonical(serializeProjection(replayCheckpointSource(serializeTransactionLog(log), "a".repeat(40), legacy[0]))), canonical(serializeProjection(replayTransactions(log)))); const state = replayTransactions(recovered); const prior = state.requests.get(log[1].request.id); const original = { nodes: log[1].request.parameters.nodes.map(({ kind, batch_trust_domain, enqueued, ...node }) => node) }; @@ -33,6 +56,29 @@ describe("checkpoint and expiration internal-state regressions", () => { assert.deepEqual(acceptedSubmissionParameters(state, context, original, prior), prior.request.parameters); }); + it("verifies fresh checkpoint source replay without trusting altered history or state digests", () => { + const log = [genesis()]; + log.push(submission(log, ["accepted"])); + const prior = "a".repeat(40); + const contents = serializeTransactionLog(log); + const root = compactTransactions(log, prior, administrator, 100)[0]; + const replayed = replayCheckpointSource(contents, prior, root); + assert.equal(canonical(serializeProjection(replayed)), canonical(serializeProjection(replayTransactions(log)))); + for (const field of ["prior_git_sha", "prior_tip", "history_sha256", "state_sha256"]) { + const changed = structuredClone(root); + changed.operations[0][field] = field === "prior_tip" ? "different-tip" : "b".repeat(field === "prior_git_sha" ? 40 : 64); + assert.throws(() => replayCheckpointSource(contents, prior, changed), /checkpoint_invalid/); + } + for (const actor of [ + { ...administrator, role: "producer" }, + { ...administrator, repository: "other/repo" }, + ]) { + assert.throws(() => replayCheckpointSource(contents, prior, { ...root, actor }), /checkpoint_invalid/); + } + const malformed = contents.replace('"version":3', '"version":3,"version":3'); + assert.throws(() => replayCheckpointSource(malformed, prior, root), /duplicate_key/); + }); + it("refreshes every public view after an isolated as-of expiration pass", () => { const log = expiringLog(); const original = canonical(log); diff --git a/actions/setup/js/work_queue_git_transport.test.cjs b/actions/setup/js/work_queue_git_transport.test.cjs index fc7bc45f7b4..f07273566b7 100644 --- a/actions/setup/js/work_queue_git_transport.test.cjs +++ b/actions/setup/js/work_queue_git_transport.test.cjs @@ -3,6 +3,7 @@ import { execFileSync } from "node:child_process"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import { performance } from "node:perf_hooks"; import { createRequire } from "node:module"; const require = createRequire(import.meta.url); @@ -20,7 +21,7 @@ afterEach(() => { for (const cleanup of cleanups.splice(0).reverse()) cleanup(); }); -function fixture() { +function fixture(options = {}) { const directory = fs.mkdtempSync(path.join(os.tmpdir(), "queue-git-test-")); cleanups.push(() => fs.rmSync(directory, { recursive: true, force: true })); const env = gitEnvironment(); @@ -30,11 +31,12 @@ function fixture() { .trim(); } git(["init", "--bare", "--quiet", "--initial-branch=main", "--template="]); - const queue = queueFixture({ count: 2, batch: 1 }); + const queue = queueFixture({ count: 2, batch: 1, ...options }); let head; let serial = 0; function store(transactions, { mode = "100644", parents = head ? [head] : [] } = {}) { - const blob = git(["hash-object", "-w", "--stdin"], transactions.map(commit => canonical(commit)).join("\n") + "\n"); + const contents = typeof transactions === "string" ? transactions : transactions.map(commit => canonical(commit)).join("\n") + "\n"; + const blob = git(["hash-object", "-w", "--stdin"], contents); const tree = git(["mktree"], `${mode} blob ${blob}\twork-queue.jsonl\n`); head = git(["commit-tree", tree, ...parents.flatMap(parent => ["-p", parent]), "-F", "-"], `Queue fixture ${++serial}`); git(["update-ref", "refs/heads/work-queue", head]); @@ -127,6 +129,81 @@ describe("Git-only work queue ledger", { timeout: 30000 }, () => { expect(f.calls).toEqual([]); }); + it("validates a synthetic live-sized nested Git history in less than three seconds from cold", async () => { + const f = fixture({ + count: 135, + granted: false, + workDefaults: { payload: { plan: "Verify the original synthetic work item without resource effects. ".repeat(80), effect_contract: { kind: "none" } } }, + }); + const retained = 1200; + const epoch = f.queue.state.policy_epoch; + let transactions = [...f.queue.transactions]; + for (let index = 0; index < retained; index++) { + const operations = [{ kind: "Control", control: "admission_paused", value: index % 2 === 0, reason: "performance_fixture" }]; + transactions.push({ + version: 3, + id: `performance-control-${index}`, + previous: transactions.at(-1).id, + request: newRequest(`performance-request-${index}`, "control", f.queue.administrator, { operations }), + actor: f.queue.administrator, + policy_epoch: epoch, + at: f.queue.at + index + 1, + operations, + }); + } + f.store(transactions); + for (let generation = 0; generation < 4; generation++) { + transactions = compactTransactions(transactions, f.head(), f.queue.administrator, f.queue.at + retained + generation); + for (let index = 0; index < 32; index++) f.store(transactions); + } + const bytes = Buffer.byteLength(transactions.map(commit => canonical(commit) + "\n").join("")); + expect(bytes).toBeGreaterThan(700 * 1024); + const start = performance.now(); + const cold = await readWorkQueueLog(f.options); + const coldMs = performance.now() - start; + expect(cold.state.works.size).toBe(135); + expect(cold.state.requests.size).toBe(f.queue.state.requests.size + retained + 4); + expect(cold.sha).toBe(f.head()); + expect(coldMs).toBeLessThan(3000); + cold.state.works.clear(); + const warmStart = performance.now(); + const warm = await readWorkQueueLog(f.options); + const warmMs = performance.now() - warmStart; + expect(warm.state.works.size).toBe(135); + expect(warmMs).toBeLessThan(3000); + expect(f.calls).toEqual([]); + console.info( + JSON.stringify({ + benchmark: "synthetic-cold-queue-validation", + works: 135, + retained_requests: retained, + checkpoint_generations: 4, + ancestry_commits: 128, + checkpoint_bytes: bytes, + cold_ms: Math.round(coldMs), + warm_ms: Math.round(warmMs), + budget_ms: 3000, + }) + ); + }); + + it.each(["duplicate-key", "truncated", "invalid-request"])("rejects a %s historical log even when the descendant checkpoint is valid", async defect => { + const f = fixture(); + const checkpoint = compactTransactions(f.queue.transactions, f.head(), f.queue.administrator, f.queue.at); + const contents = canonical(checkpoint[0]) + "\n"; + f.store(checkpoint); + if (defect === "duplicate-key") f.store(contents.replace('"version":3', '"version":3,"version":3')); + else if (defect === "truncated") f.store(contents.slice(0, -1)); + else { + const invalid = structuredClone(checkpoint[0]); + invalid.request.parameters.prior_tip = "different-tip"; + f.store([invalid]); + } + f.store(checkpoint); + await expect(readWorkQueueLog(f.options)).rejects.toThrow(); + expect(f.calls).toEqual([]); + }); + it("publishes and recovers a stable request after compaction using Git CAS only", async () => { const f = fixture(); const checkpoint = compactTransactions(f.queue.transactions, f.head(), f.queue.administrator, f.queue.at); diff --git a/actions/setup/js/work_queue_live_performance.test.cjs b/actions/setup/js/work_queue_live_performance.test.cjs index 26bea3de210..830ddef76e3 100644 --- a/actions/setup/js/work_queue_live_performance.test.cjs +++ b/actions/setup/js/work_queue_live_performance.test.cjs @@ -2,6 +2,9 @@ import { describe, expect, it } from "vitest"; import { execFileSync } from "node:child_process"; import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; import { createRequire } from "node:module"; import { digest } from "./work_queue_codec.cjs"; import { replayTransactionLog, serializeProjection } from "./work_queue_replay.cjs"; @@ -9,7 +12,7 @@ import { readWorkQueueLog } from "./work_queue_store.cjs"; const require = createRequire(import.meta.url); const { registerStaticQueueAuthentication } = require("./work_queue_auth_cache.cjs"); -const { registerQueueLedgerClient } = require("./work_queue_git_transport.cjs"); +const { createGitLedgerClient, gitEnvironment, registerQueueLedgerClient } = require("./work_queue_git_transport.cjs"); const head = process.env.GH_AW_WORK_QUEUE_PERF_LIVE_HEAD; function projectionDigest(state) { @@ -71,6 +74,36 @@ function recordedClient(repository, registered) { } describe.runIf(Boolean(head))("read-only captured live work queue performance", () => { + it("validates captured queue ancestry through the production Git transport in less than ten seconds", async () => { + const filename = process.env.GH_AW_WORK_QUEUE_PERF_LEDGER; + if (!filename || !/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(head ?? "")) throw new Error("A captured ledger and immutable Git head are required"); + const expected = replayTransactionLog(fs.readFileSync(filename, "utf8")); + const [owner, repo] = expected.repository.split("/"); + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "queue-live-performance-")); + const env = gitEnvironment(); + let ledger; + try { + const source = execFileSync("git", ["rev-parse", "--show-toplevel"], { env, encoding: "utf8" }).trim(); + execFileSync("git", ["init", "--bare", "--quiet", "--template=", directory], { env }); + execFileSync("git", ["--git-dir", directory, "fetch", "--quiet", "--no-tags", "--no-recurse-submodules", source, `${head}:refs/heads/work-queue`], { env, timeout: 120000, stdio: "pipe" }); + const githubClient = { auth: async () => ({ type: "token", token: "offline-performance-token" }) }; + ledger = createGitLedgerClient({ githubClient, owner, repo, branch: "work-queue", remote: directory }); + registerQueueLedgerClient(githubClient, ledger); + const start = performance.now(); + const cold = await readWorkQueueLog({ githubClient, owner, repo }); + const coldMs = performance.now() - start; + expect(cold.sha).toBe(head); + expect(projectionDigest(cold.state)).toBe(projectionDigest(expected)); + console.info( + JSON.stringify({ benchmark: "captured-production-git-queue", head, works: cold.state.works.size, claims: cold.state.claims.size, requests: cold.state.requests.size, cold_ms: Math.round(coldMs), budget_ms: 10000, mutations: 0 }) + ); + expect(coldMs).toBeLessThan(10000); + } finally { + ledger?.close(); + fs.rmSync(directory, { recursive: true, force: true }); + } + }, 120000); + it.each([false, true])( "verifies actual Git checkpoint history and measures repeated reads (static cache registered: %s)", async registered => { diff --git a/actions/setup/js/work_queue_replay.cjs b/actions/setup/js/work_queue_replay.cjs index 4683dda8360..1fd77f7d3d9 100644 --- a/actions/setup/js/work_queue_replay.cjs +++ b/actions/setup/js/work_queue_replay.cjs @@ -868,7 +868,7 @@ function restoreCheckpointState(checkpoint) { const invalid = () => queueError("checkpoint_invalid", "checkpoint state does not bind a valid prior projection"); if (checkpoint.previous !== null || checkpoint.actor.role !== "administrator" || checkpoint.request.kind !== "checkpoint") throw invalid(); const wire = op.state; - const snapshot = structuredClone(expandCheckpointSnapshot(wire)); + const snapshot = expandCheckpointSnapshot(structuredClone(wire)); closed( snapshot, [ @@ -1184,7 +1184,8 @@ function expandCheckpointSnapshot(wire) { if (typeof wire.requests_compressed !== "string" || !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/.test(wire.requests_compressed)) throw invalid(); try { const requestsText = inflateRawSync(Buffer.from(wire.requests_compressed, "base64"), { maxOutputLength: 64 << 20 }).toString("utf8"); - const requests = parseStrictJSON(requestsText); + // Exact canonical equality also rejects duplicate keys and lossy numbers. + const requests = JSON.parse(requestsText); if (!Array.isArray(requests) || canonical(requests) !== requestsText) throw invalid(); const { requests_compressed: _, ...projection } = wire; return { ...projection, requests }; @@ -1527,10 +1528,7 @@ function compactTransactions(transactions, priorGitSHA, actor, at) { return checkpointTransactions(transactions, priorGitSHA, actor, at, false); } -function checkpointTransactions(transactions, priorGitSHA, actor, at, legacy, retainPolicyHistory = true) { - const validated = replayTransactions(transactions); - if (priorGitSHA === undefined) return validated.transactions; - if (at === undefined) at = Math.max(Date.now(), validated.time_watermark); +function checkpointState(validated, priorGitSHA, actor, at, legacy, retainPolicyHistory) { let lastAt = validated.time_watermark; if (legacy) { if ([...validated.works.values()].some(work => work.expires_at !== undefined)) throw queueError("checkpoint_invalid", "deadline-bearing history requires the committed time watermark"); @@ -1538,9 +1536,15 @@ function checkpointTransactions(transactions, priorGitSHA, actor, at, legacy, re } if (!/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(priorGitSHA) || actor?.role !== "administrator" || actor.repository !== validated.repository || !Number.isSafeInteger(at) || at < lastAt || at > Number.MAX_SAFE_INTEGER) throw queueError("checkpoint_invalid", "checkpoint requires an administrator, valid time and prior Git commit SHA"); - const history = serializeTransactionLog(validated.transactions); + return checkpointSnapshot(validated, lastAt, retainPolicyHistory); +} + +function checkpointTransactions(transactions, priorGitSHA, actor, at, legacy, retainPolicyHistory = true, validated = replayTransactions(transactions)) { + if (priorGitSHA === undefined) return validated.transactions; + if (at === undefined) at = Math.max(Date.now(), validated.time_watermark); + const history = validated.transactions.map(transaction => canonical(transaction) + "\n").join(""); const history_sha256 = createHash("sha256").update(history, "utf8").digest("hex"); - const state = checkpointSnapshot(validated, lastAt, retainPolicyHistory); + const state = checkpointState(validated, priorGitSHA, actor, at, legacy, retainPolicyHistory); const state_sha256 = digest(state); const requests_compressed = deflateRawSync(Buffer.from(canonical(state.requests), "utf8")).toString("base64"); const { requests: _, ...projection } = state; @@ -1575,12 +1579,13 @@ function prepareCheckpoint(source, priorGitSHA, actor, at) { // Only history verification may reproduce the pre-expiration final-timestamp // hash. New checkpoints always use the committed watermark. function reconstructCheckpoint(transactions, priorGitSHA, actor, at, expectedStateSHA) { - const root = causalChain(transactions)[0]; + const validated = replayTransactions(transactions); + const root = validated.transactions[0]; const administrator = actor ?? { role: "administrator", principal: root.actor.principal, repository: root.actor.repository }; for (const retainPolicyHistory of [true, false]) { for (const legacy of [false, true]) { try { - const checkpoint = checkpointTransactions(transactions, priorGitSHA, administrator, at, legacy, retainPolicyHistory); + const checkpoint = checkpointTransactions(transactions, priorGitSHA, administrator, at, legacy, retainPolicyHistory, validated); if (checkpoint[0].operations[0].state_sha256 === expectedStateSHA) return checkpoint; } catch (error) { if (error?.code !== "checkpoint_invalid") throw error; @@ -1590,6 +1595,30 @@ function reconstructCheckpoint(transactions, priorGitSHA, actor, at, expectedSta throw queueError("checkpoint_invalid", "checkpoint does not match its Git parent history"); } +function replayCheckpointSource(contents, priorGitSHA, root) { + const state = replayTransactionLog(contents); + const checkpoint = root.operations[0]; + const invalid = () => queueError("checkpoint_invalid", "checkpoint does not match its Git parent history"); + if ( + priorGitSHA !== checkpoint.prior_git_sha || + state.tip !== checkpoint.prior_tip || + createHash("sha256") + .update(state.transactions.map(transaction => canonical(transaction) + "\n").join(""), "utf8") + .digest("hex") !== checkpoint.history_sha256 + ) + throw invalid(); + for (const retainPolicyHistory of [true, false]) { + for (const legacy of [false, true]) { + try { + if (digest(checkpointState(state, priorGitSHA, root.actor, root.at, legacy, retainPolicyHistory)) === checkpoint.state_sha256) return state; + } catch (error) { + if (error?.code !== "checkpoint_invalid") throw error; + } + } + } + throw invalid(); +} + function appendCommit(transactions, commit) { log.debug("commit.append.start", { transactions: transactions.length }); validateCommit(commit); @@ -1690,6 +1719,7 @@ module.exports = { planDispatchWithObservations, prepareCheckpoint, reconstructCheckpoint, + replayCheckpointSource, proposedCommitId, planNext, poolIdentity, diff --git a/actions/setup/js/work_queue_store.cjs b/actions/setup/js/work_queue_store.cjs index 33aecb63520..4b224565131 100644 --- a/actions/setup/js/work_queue_store.cjs +++ b/actions/setup/js/work_queue_store.cjs @@ -26,6 +26,7 @@ const { reconstructCheckpoint, proposedCommitId, replayTransactionLog, + replayCheckpointSource, replayTransactions, validateClaimAuthority, validateRequest, @@ -186,9 +187,10 @@ async function readVerifiedQueueAtSHA(options, sha, verify = true, recover = tru * @param {string} sha * @param {Record | undefined} [request] * @param {(code: string, message: string) => Error} [onInvalid] + * @param {Map | undefined} [knownPrefixes] * @throws {Error} If queue contents are invalid, using onInvalid when provided. */ -async function readQueueContentsAtSHA({ githubClient, owner, repo }, sha, request = undefined, onInvalid = undefined) { +async function readQueueContentsAtSHA({ githubClient, owner, repo }, sha, request = undefined, onInvalid = undefined, knownPrefixes = undefined) { /** * @type {(code: string, message: string) => never} * @throws {Error} Always throws the error from onInvalid or queueError. @@ -207,6 +209,8 @@ async function readQueueContentsAtSHA({ githubClient, owner, repo }, sha, reques if (entries.some(entry => LEGACY_LOGS.includes(entry.path))) fail("unsupported_protocol", "legacy transaction log cannot be adopted"); const logs = entries.filter(entry => entry.path === WORK_QUEUE_LOG_PATH); if (logs.length !== 1 || logs[0].type !== "blob" || typeof logs[0].sha !== "string" || logs[0].mode !== "100644") fail("ledger_invalid", "existing queue must contain exactly one regular canonical log"); + const known = knownPrefixes?.get(logs[0].sha); + if (known !== undefined) return { treeSha, contents: known, prefixVerified: true }; const blob = await githubClient.rest.git.getBlob({ owner, repo, file_sha: logs[0].sha, ...requestOptions }); if (blob?.data?.encoding !== "base64" || typeof blob.data.content !== "string") fail("ledger_invalid", "queue blob encoding is unsupported"); const encoded = blob.data.content.replace(/\s/g, ""); @@ -348,7 +352,7 @@ async function replayVerifiedQueueLog(options, sha, contents, verify = true, rec checkpointRecovered = true; } if (state.transactions.some(transaction => transaction.actor.repository.toLowerCase() !== options.repository.full_name.toLowerCase())) throw queueError("actor_unauthorized", "queue log contains an actor from another repository"); - if (verify && !checkpointRecovered && state.transactions[0]?.operations[0]?.kind === "Checkpoint") await verifyCheckpointGitHistory(options, sha, state.transactions[0]); + if (verify && !checkpointRecovered && state.transactions[0]?.operations[0]?.kind === "Checkpoint") await verifyCheckpointGitHistory(options, sha, state.transactions); return { state, checkpointRecovered }; } @@ -371,14 +375,14 @@ async function recoverDeploymentCheckpoint(options, sha, contents, failure) { validateExtendingPrefix(original.transactions, repaired, "checkpoint recovery cannot change accepted history"); const state = replayTransactions(repaired); if (canonical(damaged.operations[0].state.deployments) !== canonical(serializeDeployments(state))) throw failure; - await verifyCheckpointGitHistory(options, parent, root); + await verifyCheckpointGitHistory(options, parent, original.transactions); const message = "Work queue: recovered aliased checkpoint deployments from its verified Git parent; the next checked publication will persist the unchanged checkpoint and retained requests."; if (typeof coreApi?.warning === "function") coreApi.warning(message); else coreApi?.info(message); return state; } -async function verifyCheckpointGitHistory(options, head, root) { +async function verifyCheckpointGitHistory(options, head, transactions) { options = { ...options, githubClient: queueLedgerClient(options.githubClient, options.owner, options.repo, options.branch) }; const { githubClient, owner, repo } = options; const invalid = () => queueError("checkpoint_invalid", "checkpoint does not match its Git parent history"); @@ -388,9 +392,14 @@ async function verifyCheckpointGitHistory(options, head, root) { if (!scope || cached?.scope !== scope) verifiedCheckpointHistory.delete(githubClient); const verified = cached && scope && cached.scope === scope ? cached.proofs : new Set(); const pending = new Set(); + let root = transactions[0]; verification: while (root?.operations?.[0]?.kind === "Checkpoint") { const checkpoint = root.operations[0]; const checkpointDigest = digest(root); + let lines = transactions.map(transaction => canonical(transaction) + "\n"); + const rootBytes = Buffer.byteLength(lines[0], "utf8"); + let validatedPrefix = lines.join(""); + let prefixes = queueGitScope(githubClient) ? canonicalGitPrefixes(lines, head.length) : undefined; let current = head; for (;;) { // Git reads recheck repository access; other transports require static auth. @@ -403,9 +412,36 @@ async function verifyCheckpointGitHistory(options, head, root) { const parents = commit?.data?.parents; if (!Array.isArray(parents) || parents.length !== 1 || !parents[0]?.sha || parents[0].sha === current) throw invalid(); const parent = parents[0].sha; - const previous = await readVerifiedQueueAtSHA({ ...options, coreApi: undefined }, parent, false); + const cachedParent = verifiedQueueContents.get(githubClient); + const immutable = + cachedParent && scope && cachedParent.scope === scope && cachedParent.repository === options.repository.full_name.toLowerCase() && cachedParent.sha === parent + ? cachedParent + : await readQueueContentsAtSHA(options, parent, undefined, undefined, prefixes); + const { contents } = immutable; + // Exact canonical causal prefixes inherit validation, not mutable projections. + if (("prefixVerified" in immutable && immutable.prefixVerified) || (contents.endsWith("\n") && Buffer.byteLength(contents, "utf8") >= rootBytes && validatedPrefix.startsWith(contents))) { + current = parent; + continue; + } + if (parent === checkpoint.prior_git_sha) { + try { + const state = replayCheckpointSource(contents, parent, root); + head = parent; + transactions = state.transactions; + root = transactions[0]; + break; + } catch (error) { + // Preserve the existing deployment-checkpoint recovery path. + if (error?.code !== "checkpoint_invalid") throw error; + } + } + const { state } = await replayVerifiedQueueLog({ ...options, coreApi: undefined }, parent, contents, false); + const previous = { transactions: state.transactions, state }; if (previous.transactions[0]?.id === root.id) { if (canonical(previous.transactions[0]) !== canonical(root)) throw invalid(); + lines = previous.transactions.map(transaction => canonical(transaction) + "\n"); + validatedPrefix = lines.join(""); + prefixes = queueGitScope(githubClient) ? canonicalGitPrefixes(lines, head.length) : undefined; current = parent; continue; } @@ -419,6 +455,7 @@ async function verifyCheckpointGitHistory(options, head, root) { throw invalid(); if (previous.state.tip !== checkpoint.prior_tip || reconstructCheckpoint(previous.transactions, parent, root.actor, root.at, checkpoint.state_sha256)[0].operations[0].state_sha256 !== checkpoint.state_sha256) throw invalid(); head = parent; + transactions = previous.transactions; root = previous.transactions[0]; break; } @@ -435,6 +472,25 @@ async function verifyCheckpointGitHistory(options, head, root) { verifiedCheckpointHistory.set(githubClient, { scope, proofs: verified }); } +function canonicalGitPrefixes(lines, oidLength) { + if (lines.length > 256 || ![40, 64].includes(oidLength)) return undefined; + const contents = lines.join(""); + const bytes = Buffer.from(contents, "utf8"); + const prefixes = new Map(); + let offset = 0; + let end = 0; + for (const line of lines) { + offset += Buffer.byteLength(line, "utf8"); + end += line.length; + const oid = createHash(oidLength === 40 ? "sha1" : "sha256") + .update(`blob ${offset}\0`, "utf8") + .update(bytes.subarray(0, offset)) + .digest("hex"); + prefixes.set(oid, contents.slice(0, end)); + } + return prefixes; +} + function stableRequestResult(current, request, actor) { const commit = current.state.requests.get(request.id); if (!commit) return null; diff --git a/docs/src/content/docs/specs/work-queue-specification.md b/docs/src/content/docs/specs/work-queue-specification.md index 650f5bc960e..8af1db91f0b 100644 --- a/docs/src/content/docs/specs/work-queue-specification.md +++ b/docs/src/content/docs/specs/work-queue-specification.md @@ -887,6 +887,24 @@ recomputed state digest, including earlier checkpoints. It MUST NOT depend on API pagination, API caches, Issue mirrors or native run queries. This changes transport, not the version-3 wire format or the checkpoint trust boundary. +While walking single-parent Git history, an exact newline-terminated canonical +causal prefix of an already validated log may reuse its transaction validation. +For Git transport, the canonical prefix's complete Git blob object identity +may establish that byte equality without reading the same large payload again. +Every Git parent is still checked, and every checkpoint boundary is still +reconstructed from a freshly replayed source snapshot; verification need not +compress and replay a throwaway replacement checkpoint. Noncanonical, changed +or malformed historical logs require full validation; a matching checkpoint ID +alone is never sufficient. No mutable +projection is shared between reads. The synthetic Git-history benchmark includes +135 Work items, 1,200 retained requests, four checkpoint generations and 128 +ancestry commits, with a +three-second cold-validation budget excluding fixture construction and without +hosted network latency. This budget does not extend durable delivery deadlines. +The optional captured-history benchmark checks the exact immutable head and +replayed projection through production Git transport, including its cold +object transfer and integrity checks, with a ten-second local-read ceiling. + Hosted identity, approved worker configuration, native launch/termination and effect delivery remain separate evidence boundaries and MAY require APIs. Preparing a branchless repository's default branch is likewise a hosting From 5a80d7f34f30c1f6343cfcd4797e36e0bfaf94a9 Mon Sep 17 00:00:00 2001 From: pelikhan Date: Thu, 5 Nov 2026 23:31:54 -0800 Subject: [PATCH 2/2] test(work-queue): cover full cold Git history Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../js/work_queue_git_transport.test.cjs | 118 +++++++++++++----- .../docs/specs/work-queue-specification.md | 5 + 2 files changed, 95 insertions(+), 28 deletions(-) diff --git a/actions/setup/js/work_queue_git_transport.test.cjs b/actions/setup/js/work_queue_git_transport.test.cjs index f07273566b7..348f9902526 100644 --- a/actions/setup/js/work_queue_git_transport.test.cjs +++ b/actions/setup/js/work_queue_git_transport.test.cjs @@ -11,7 +11,7 @@ const { createGitLedgerClient, gitEnvironment, queueLedgerClient, registerQueueL const { wrapGithubClient } = require("./work_queue_issue_pacing.cjs"); const { readWorkQueueLog, publishWorkQueueRequest } = require("./work_queue_store.cjs"); const { readCheckedQueue, writeCheckedFiles } = require("./work_queue_checked_transport.cjs"); -const { compactTransactions, newRequest, serializeTransactionLog } = require("./work_queue_replay.cjs"); +const { compactTransactions, newRequest, replayTransactions, serializeProjection, serializeTransactionLog } = require("./work_queue_replay.cjs"); const { queueFixture } = require("./work_queue_lifecycle.test_helpers.cjs"); const { context } = require("./work_queue_test_helpers.cjs"); const { canonical } = require("./work_queue_codec.cjs"); @@ -42,6 +42,20 @@ function fixture(options = {}) { git(["update-ref", "refs/heads/work-queue", head]); return head; } + function storeHistory(transactions, count) { + const contents = transactions.map(commit => canonical(commit) + "\n").join(""); + const input = [`blob\nmark :1\ndata ${Buffer.byteLength(contents)}\n${contents}\n`]; + for (let index = 0; index < count; index++) { + const message = `Queue fixture ${++serial}`; + input.push( + `commit refs/heads/work-queue\nmark :${index + 2}\ncommitter Queue fixture ${queue.at} +0000\ndata ${Buffer.byteLength(message)}\n${message}\nfrom ${index ? `:${index + 1}` : head}\nM 100644 :1 work-queue.jsonl\n\n` + ); + } + input.push("done\n"); + git(["fast-import", "--quiet"], input.join("")); + head = git(["rev-parse", "refs/heads/work-queue"]); + return head; + } store(queue.transactions); const calls = []; const networkCalls = []; @@ -60,7 +74,41 @@ function fixture(options = {}) { const ledger = createGitLedgerClient({ githubClient, owner: "owner", repo: "repo", branch: "work-queue", remote: directory }); registerQueueLedgerClient(githubClient, ledger); cleanups.push(() => ledger.close()); - return { queue, githubClient, ledger, calls, networkCalls, store, git, options: { githubClient, owner: "owner", repo: "repo" }, head: () => head }; + return { queue, githubClient, ledger, calls, networkCalls, store, storeHistory, git, options: { githubClient, owner: "owner", repo: "repo" }, head: () => head }; +} + +function performanceHistory({ generations = 4, commitsPerCheckpoint = 32, bound = false, corruptOldest = false } = {}) { + const f = fixture({ + count: 135, + granted: bound, + bound, + workDefaults: { payload: { plan: "Verify the original synthetic work item without resource effects. ".repeat(80), effect_contract: { kind: "none" } } }, + }); + const retained = 1200; + const initial = f.queue.state; + let transactions = [...f.queue.transactions]; + for (let index = 0; index < retained; index++) { + const operations = [{ kind: "Control", control: "admission_paused", value: index % 2 === 0, reason: "performance_fixture" }]; + transactions.push({ + version: 3, + id: `performance-control-${index}`, + previous: transactions.at(-1).id, + request: newRequest(`performance-request-${index}`, "control", f.queue.administrator, { operations }), + actor: f.queue.administrator, + policy_epoch: initial.policy_epoch, + at: f.queue.at + index + 1, + operations, + }); + } + f.store(transactions); + for (let generation = 0; generation < generations; generation++) { + const source = transactions; + transactions = compactTransactions(source, f.head(), f.queue.administrator, f.queue.at + retained + generation); + // Preserve valid logical snapshots while breaking the oldest checkpoint's Git-parent binding. + if (generation === 0 && corruptOldest) f.store(source, { parents: [] }); + f.storeHistory(transactions, commitsPerCheckpoint); + } + return { f, transactions, retained, generations, commitsPerCheckpoint, initial }; } describe("Git-only work queue ledger", { timeout: 30000 }, () => { @@ -130,32 +178,7 @@ describe("Git-only work queue ledger", { timeout: 30000 }, () => { }); it("validates a synthetic live-sized nested Git history in less than three seconds from cold", async () => { - const f = fixture({ - count: 135, - granted: false, - workDefaults: { payload: { plan: "Verify the original synthetic work item without resource effects. ".repeat(80), effect_contract: { kind: "none" } } }, - }); - const retained = 1200; - const epoch = f.queue.state.policy_epoch; - let transactions = [...f.queue.transactions]; - for (let index = 0; index < retained; index++) { - const operations = [{ kind: "Control", control: "admission_paused", value: index % 2 === 0, reason: "performance_fixture" }]; - transactions.push({ - version: 3, - id: `performance-control-${index}`, - previous: transactions.at(-1).id, - request: newRequest(`performance-request-${index}`, "control", f.queue.administrator, { operations }), - actor: f.queue.administrator, - policy_epoch: epoch, - at: f.queue.at + index + 1, - operations, - }); - } - f.store(transactions); - for (let generation = 0; generation < 4; generation++) { - transactions = compactTransactions(transactions, f.head(), f.queue.administrator, f.queue.at + retained + generation); - for (let index = 0; index < 32; index++) f.store(transactions); - } + const { f, transactions, retained } = performanceHistory(); const bytes = Buffer.byteLength(transactions.map(commit => canonical(commit) + "\n").join("")); expect(bytes).toBeGreaterThan(700 * 1024); const start = performance.now(); @@ -187,6 +210,45 @@ describe("Git-only work queue ledger", { timeout: 30000 }, () => { ); }); + it.each([false, true])("verifies the full cold Git history, including its oldest checkpoint (corrupted: %s)", async corruptOldest => { + const { f, transactions, retained, generations, commitsPerCheckpoint, initial } = performanceHistory({ generations: 70, commitsPerCheckpoint: 21, bound: true, corruptOldest }); + expect(Number(f.git(["rev-list", "--count", f.head()]))).toBe(generations * commitsPerCheckpoint + (corruptOldest ? 1 : 2)); + expect(f.networkCalls).toEqual([]); + expect(initial.claims.size).toBeGreaterThan(0); + const expected = canonical(serializeProjection(replayTransactions(transactions))); + const start = performance.now(); + let cold; + if (corruptOldest) { + await expect(readWorkQueueLog(f.options)).rejects.toMatchObject({ code: "checkpoint_invalid" }); + } else { + cold = await readWorkQueueLog(f.options); + } + const coldMs = performance.now() - start; + if (!corruptOldest) { + expect(cold.sha).toBe(f.head()); + expect(canonical(serializeProjection(cold.state))).toBe(expected); + expect(cold.transactions).toEqual(transactions); + expect(cold.state.requests.size).toBe(initial.requests.size + retained + generations); + expect(cold.state.claims.size).toBe(initial.claims.size); + } + expect(f.networkCalls).toHaveLength(2); + expect(f.calls).toEqual([]); + console.info( + JSON.stringify({ + benchmark: "full-git-history-verification", + works: 135, + claims: initial.claims.size, + retained_requests: retained, + checkpoint_generations: generations, + ancestry_commits: generations * commitsPerCheckpoint, + corrupt_oldest: corruptOldest, + cold_ms: Math.round(coldMs), + budget_ms: 10000, + }) + ); + expect(coldMs).toBeLessThan(10000); + }); + it.each(["duplicate-key", "truncated", "invalid-request"])("rejects a %s historical log even when the descendant checkpoint is valid", async defect => { const f = fixture(); const checkpoint = compactTransactions(f.queue.transactions, f.head(), f.queue.administrator, f.queue.at); diff --git a/docs/src/content/docs/specs/work-queue-specification.md b/docs/src/content/docs/specs/work-queue-specification.md index 8af1db91f0b..0e18e19aea7 100644 --- a/docs/src/content/docs/specs/work-queue-specification.md +++ b/docs/src/content/docs/specs/work-queue-specification.md @@ -901,6 +901,11 @@ projection is shared between reads. The synthetic Git-history benchmark includes ancestry commits, with a three-second cold-validation budget excluding fixture construction and without hosted network latency. This budget does not extend durable delivery deadlines. +An always-on full-history test covers 70 checkpoint generations and 1,470 +ancestry commits with retained Claims and requests. It checks the exact head and +complete replayed projection through cold production Git transport, and rejects +a broken oldest-checkpoint Git-parent binding despite valid newer snapshots. +Both cases have a ten-second local-read ceiling; fixture construction is excluded. The optional captured-history benchmark checks the exact immutable head and replayed projection through production Git transport, including its cold object transfer and integrity checks, with a ten-second local-read ceiling.