Apply queued actions through a checkpoint

packages/backend-utils/__tests__/gatekeeper-action.test.ts

@@ -0,0 +1,157 @@
+import { describe, expect, it } from "vitest";
+import {
+  InvalidationLog,
+  SerialTaskQueue,
+  displayReason,
+  validateApplyThroughArgs,
+} from "../src/gatekeeper-action";
+
+type Kv = ConstructorParameters<typeof InvalidationLog>[0];
+
+// Minimal in-memory KV matching the slice of the DO storage API the log uses.
+function makeKv(): Kv {
+  const map = new Map<string, unknown>();
+  return {
+    put: (k: string, v: unknown) => void map.set(k, v),
+    delete: (k: string) => void map.delete(k),
+    list: <T>({ prefix }: { prefix: string }) =>
+      [...map.entries()].filter(([k]) => k.startsWith(prefix)) as [string, T][],
+  } as unknown as Kv;
+}
+
+describe("validateApplyThroughArgs", () => {
+  it("deduplicates in-range vetoes", () => {
+    expect(validateApplyThroughArgs(5, [2, 2, 5])).toEqual(new Set([2, 5]));
+  });
+
+  it("rejects a non-positive or non-integer frontier", () => {
+    expect(() => validateApplyThroughArgs(0, [])).toThrow("Invalid action ID");
+    expect(() => validateApplyThroughArgs(1.5, [])).toThrow("Invalid action ID");
+  });
+
+  it("rejects vetoes above the frontier or out of the integer range", () => {
+    expect(() => validateApplyThroughArgs(3, [4])).toThrow("Invalid veto action ID");
+    expect(() => validateApplyThroughArgs(3, [0])).toThrow("Invalid veto action ID");
+  });
+});
+
+describe("InvalidationLog", () => {
+  it("reports only entries attributed to the requested vetoes, ascending", () => {
+    const log = new InvalidationLog(makeKv());
+    log.record(10, 3);
+    log.record(4, 3);
+    log.record(7, 5);
+
+    expect(log.attributedTo(new Set([3]))).toEqual([
+      { action: 4, invalidatedBy: 3 },
+      { action: 10, invalidatedBy: 3 },
+    ]);
+  });
+
+  it("prunes entries below the pending floor but keeps the current request's vetoes", () => {
+    const log = new InvalidationLog(makeKv());
+    log.record(4, 3);
+    log.record(7, 5);
+
+    log.prune(new Set([3]), 4);
+
+    expect(log.attributedTo(new Set([3, 5]))).toEqual([
+      { action: 4, invalidatedBy: 3 },
+      { action: 7, invalidatedBy: 5 },
+    ]);
+
+    log.prune(new Set(), Infinity);
+
+    expect(log.attributedTo(new Set([3, 5]))).toEqual([]);
+  });
+});
+
+describe("displayReason", () => {
+  it("passes through an Error with a message", () => {
+    const error = new Error("page was deleted upstream");
+    expect(displayReason(error, "fallback")).toBe(error);
+  });
+
+  it("wraps non-Error throws in the fallback text", () => {
+    expect(displayReason("oops", "Vendor could not apply this action").message)
+      .toBe("Vendor could not apply this action: oops");
+  });
+});
+
+// This package deliberately avoids the full Node type environment (see node-async-hooks.d.ts).
+// Tests run under vitest on Node, so type the small `process` surface used here locally.
+const nodeProcess = (globalThis as Record<string, unknown>)["process"] as {
+  on(event: "unhandledRejection", handler: (reason: unknown) => void): void;
+  off(event: "unhandledRejection", handler: (reason: unknown) => void): void;
+};
+
+describe("SerialTaskQueue", () => {
+  it("runs operations one at a time in submission order", async () => {
+    const queue = new SerialTaskQueue();
+    const events: string[] = [];
+    let release!: () => void;
+    const blocked = new Promise<void>(resolve => { release = resolve; });
+
+    const first = queue.run(async () => {
+      events.push("first:start");
+      await blocked;
+      events.push("first:end");
+      return 1;
+    });
+    const second = queue.run(() => {
+      events.push("second");
+      return 2;
+    });
+
+    await Promise.resolve();
+    expect(events).toEqual(["first:start"]);
+    release();
+    await expect(Promise.all([first, second])).resolves.toEqual([1, 2]);
+    expect(events).toEqual(["first:start", "first:end", "second"]);
+  });
+
+  it("continues after an operation rejects", async () => {
+    const queue = new SerialTaskQueue();
+    const failed = queue.run(() => { throw new Error("failed"); });
+    const recovered = queue.run(() => "recovered");
+
+    await expect(failed).rejects.toThrow("failed");
+    await expect(recovered).resolves.toBe("recovered");
+  });
+
+  it("preserves order across an asynchronous rejection", async () => {
+    const queue = new SerialTaskQueue();
+    const events: string[] = [];
+
+    const failed = queue.run(async () => {
+      events.push("first");
+      await Promise.resolve();
+      throw new Error("async failure");
+    });
+    const second = queue.run(() => {
+      events.push("second");
+      return "ok";
+    });
+
+    await expect(failed).rejects.toThrow("async failure");
+    await expect(second).resolves.toBe("ok");
+    expect(events).toEqual(["first", "second"]);
+  });
+
+  it("does not leak an unhandled rejection from its internal chain", async () => {
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown) => void unhandled.push(reason);
+    nodeProcess.on("unhandledRejection", onUnhandled);
+    try {
+      const queue = new SerialTaskQueue();
+      // A synchronous throw rejects the queue's chained promises directly, so a missing rejection
+      // handler on the internal tail promise would surface here as an unhandled rejection.
+      await expect(queue.run(() => { throw new Error("boom"); })).rejects.toThrow("boom");
+      // Unhandled rejections are reported asynchronously, after the microtask queue drains.
+      await new Promise(resolve => setTimeout(resolve, 0));
+      expect(unhandled).toEqual([]);
+    } finally {
+      nodeProcess.off("unhandledRejection", onUnhandled);
+    }
+  });
+});

packages/backend-utils/package.json

@@ -19,6 +19,10 @@
     "./error-reporting": {
       "types": "./src/error-reporting.ts",
       "import": "./src/error-reporting.ts"
+    },
+    "./gatekeeper-action": {
+      "types": "./src/gatekeeper-action.ts",
+      "import": "./src/gatekeeper-action.ts"
     }
   },
   "scripts": {

packages/backend-utils/src/gatekeeper-action.ts

@@ -0,0 +1,88 @@
+// Shared helpers for gatekeepers implementing the `Gatekeeper` action contract. They cover the
+// obligations every action-queueing gatekeeper repeats: serializing resolution methods, argument
+// validation, durable attribution of veto-cascade invalidations (so repeated requests can
+// re-report them), and the display-safe `stopped.reason` error.
+
+type Kv = DurableObjectStorage["kv"];
+
+/** Runs asynchronous operations sequentially in submission order. */
+export class SerialTaskQueue {
+  #tail: Promise<void> = Promise.resolve();
+
+  /** Enqueues an operation without allowing a rejection to block later operations. */
+  run<T>(operation: () => T | Promise<T>): Promise<T> {
+    const result = this.#tail.then(operation);
+    this.#tail = result.then(() => undefined, () => undefined);
+    return result;
+  }
+}
+
+/** One `ApplyActionsThroughResult.invalidatedByVeto` entry. */
+export type VetoInvalidation = { action: number, invalidatedBy: number };
+
+/**
+ * Validate an `applyActionsThrough(actionId, vetoes)` request before touching any state: the
+ * frontier must be a positive integer and every veto must be a positive integer at or below it.
+ * Returns the deduplicated veto set.
+ */
+export function validateApplyThroughArgs(actionId: number, vetoes: number[]): Set<number> {
+  if (!Number.isSafeInteger(actionId) || actionId < 1) throw new TypeError("Invalid action ID.");
+  const result = new Set<number>();
+  for (const veto of vetoes) {
+    if (!Number.isSafeInteger(veto) || veto < 1 || veto > actionId) {
+      throw new TypeError("Invalid veto action ID.");
+    }
+    result.add(veto);
+  }
+  return result;
+}
+
+/**
+ * Durable record of which veto invalidated which cascade-deleted action, kept in its own KV
+ * keyspace so a gatekeeper that deletes rejected records can still satisfy the contract's
+ * requirement that a repeated request re-report invalidations attributable to its vetoes.
+ */
+export class InvalidationLog {
+  #kv: Kv;
+
+  constructor(kv: Kv) {
+    this.#kv = kv;
+  }
+
+  /** Record that the veto of `vetoedBy` invalidated `invalidatedId`, for later re-reporting. */
+  record(invalidatedId: number, vetoedBy: number): void {
+    this.#kv.put(`invalidation:${invalidatedId}`, vetoedBy);
+  }
+
+  /** Persisted invalidations attributed to any of the given vetoed action IDs, ascending. */
+  attributedTo(vetoes: Set<number>): VetoInvalidation[] {
+    return [...this.#kv.list<number>({ prefix: "invalidation:" })]
+      .map(([key, invalidatedBy]) => ({ action: Number(key.slice("invalidation:".length)), invalidatedBy }))
+      .filter(entry => vetoes.has(entry.invalidatedBy))
+      .toSorted((a, b) => a.action - b.action);
+  }
+
+  /**
+   * Drop entries no future request can attribute: a veto is only ever re-sent while some action at
+   * or below it is still undecided, so entries whose veto precedes every remaining undecided
+   * action (`pendingFloor`, `Infinity` when none remain) are unreachable. `keep` protects the
+   * current request's vetoes.
+   */
+  prune(keep: Set<number>, pendingFloor: number): void {
+    for (const [key, vetoedBy] of this.#kv.list<number>({ prefix: "invalidation:" })) {
+      if (vetoedBy < pendingFloor && !keep.has(vetoedBy)) this.#kv.delete(key);
+    }
+  }
+}
+
+/**
+ * The error a gatekeeper should report in `stopped.reason`. Only its `message` survives the RPC
+ * hop to the overseer, so it must stand alone as text the user can act on; apply errors already
+ * carry a specific, display-safe message, exactly as the legacy single-action path surfaced them.
+ * `fallback` describes the failure generically for non-Error throws with no message of their own.
+ */
+export function displayReason(error: unknown, fallback: string): Error {
+  return error instanceof Error && error.message
+    ? error
+    : new Error(`${fallback}: ${String(error)}`);
+}

packages/backend-utils/tsconfig.json

@@ -2,7 +2,7 @@
   "extends": "../../tsconfig.json",
   "compilerOptions": {
     "target": "ES2022",
-    "lib": ["ES2022"],
+    "lib": ["ES2023"],
     "module": "ESNext",
     "moduleResolution": "bundler",
     "types": ["@cloudflare/workers-types/experimental", "./src/node-async-hooks.d.ts"]