Speed up work queue checkpoint verification

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,13 +48,37 @@ 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) };
     const context = { ...prior.actor, authenticated: true, roles: [prior.actor.role], created_at: log[1].at };
     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);

actions/setup/js/work_queue_git_transport.test.cjs

@@ -3,14 +3,15 @@ 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);
 const { createGitLedgerClient, gitEnvironment, queueLedgerClient, registerQueueLedgerClient } = require("./work_queue_git_transport.cjs");
 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");
@@ -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,16 +31,31 @@ 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]);
     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-fixture@example.invalid> ${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 = [];
@@ -58,7 +74,41 @@ function fixture() {
   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 }, () => {
@@ -127,6 +177,95 @@ 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, transactions, retained } = performanceHistory();
+    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([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);
+    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);

actions/setup/js/work_queue_live_performance.test.cjs

@@ -2,14 +2,17 @@
 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";
 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 => {