Migrate Google and Spotify action resolution

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

@@ -1,5 +1,82 @@
 import { describe, expect, it } from "vitest";
-import { SerialTaskQueue } from "../src/gatekeeper-action";
+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.

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

@@ -1,4 +1,9 @@
-// Shared helpers for gatekeepers implementing the `Gatekeeper` action contract.
+// 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 {
@@ -11,3 +16,73 @@ export class SerialTaskQueue {
     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"]

packages/gatekeeper-confluence/__tests__/apply.test.ts

@@ -1,8 +1,10 @@
 import { describe, expect, it } from "vitest";
 import {
   ConfluenceStore,
+  applyStoredActionsThrough,
   applyStoredAction,
   revertStoredAction,
+  stageAction,
   type ConfluenceAction,
 } from "../src/confluence-actions";
 import type { ConfluenceApi } from "../src/confluence-api";
@@ -114,6 +116,205 @@ describe("applyStoredAction", () => {
   });
 });
 
+describe("applyStoredActionsThrough", () => {
+  it("skips sparse IDs and stops at the first failed action", async () => {
+    const { api, calls } = makeApi();
+    const store = storeWith(api);
+    const first = stage(store, { type: "addComment", contentId: "first", text: "one" });
+    const hole = stage(store, { type: "addComment", contentId: "hole", text: "two" });
+    const failed = stage(store, { type: "addComment", contentId: "failed", text: "three" });
+    const later = stage(store, { type: "addComment", contentId: "later", text: "four" });
+    store.deleteAction(hole);
+    api.addComment = async (id: string, _storage: string, type: string) => {
+      calls.addComment.push({ id, type });
+      if (id === "failed") throw new Error("safe failure");
+      return { id: `comment-${id}` };
+    };
+
+    const result = await applyStoredActionsThrough(store, later, []);
+
+    expect(calls.addComment.map(call => call.id)).toEqual(["first", "failed"]);
+    expect(store.getAction(first)?.state).toBe("applied");
+    expect(store.getAction(failed)?.state).toBe("pending");
+    expect(store.getAction(later)?.state).toBe("pending");
+    // The specific apply error is passed through so the user can resolve the problem.
+    expect(result.stopped).toMatchObject({ at: failed, reason: expect.any(Error) });
+    expect(result.stopped?.reason.message).toBe("safe failure");
+  });
+
+  it("does not re-apply earlier actions when retried after a stop", async () => {
+    const { api, calls } = makeApi();
+    const store = storeWith(api);
+    const first = stage(store, { type: "addComment", contentId: "first", text: "one" });
+    const flaky = stage(store, { type: "addComment", contentId: "flaky", text: "two" });
+    let failOnce = true;
+    api.addComment = async (id: string, _storage: string, type: string) => {
+      calls.addComment.push({ id, type });
+      if (id === "flaky" && failOnce) {
+        failOnce = false;
+        throw new Error("temporarily unavailable");
+      }
+      return { id: `comment-${id}` };
+    };
+
+    const stopped = await applyStoredActionsThrough(store, flaky, []);
+    const retry = await applyStoredActionsThrough(store, flaky, []);
+
+    expect(stopped.stopped).toMatchObject({ at: flaky });
+    expect(retry).toEqual({});
+    expect(calls.addComment.map(call => call.id)).toEqual(["first", "flaky", "flaky"]);
+    expect(store.getAction(first)?.state).toBe("applied");
+    expect(store.getAction(flaky)?.state).toBe("applied");
+  });
+
+  it("never applies a staged action whose submission has not completed", async () => {
+    const { api, calls } = makeApi();
+    const store = storeWith(api);
+    const staged = store.nextActionId();
+    store.putAction({
+      id: staged, action: { type: "addComment", contentId: "staged", text: "early" },
+      state: "staged", submittedAt: staged,
+    });
+    const pending = stage(store, { type: "addComment", contentId: "later", text: "late" });
+
+    const result = await applyStoredActionsThrough(store, pending, []);
+
+    expect(result).toEqual({});
+    expect(calls.addComment.map(call => call.id)).toEqual(["later"]);
+    expect(store.getAction(staged)?.state).toBe("staged");
+  });
+
+  it("persists transitive invalidations and reports them again on retry", async () => {
+    const { api, calls } = makeApi();
+    const store = storeWith(api);
+    const root = stage(store, {
+      type: "createContent", provisionalId: "~root", kind: "page",
+      parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current",
+    });
+    const edit = stage(store, {
+      type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root",
+    });
+    const child = stage(store, {
+      type: "createContent", provisionalId: "~child", kind: "page",
+      parent: { type: "page", parentId: "~root", spaceKey: "ENG" }, title: "Child", status: "current",
+    });
+    const childEdit = stage(store, {
+      type: "addComment", contentId: "~child", text: "Comment",
+    });
+
+    const first = await applyStoredActionsThrough(store, root, [root]);
+    const retry = await applyStoredActionsThrough(store, root, [root]);
+
+    expect(first.invalidatedByVeto).toEqual([
+      { action: edit, invalidatedBy: root },
+      { action: child, invalidatedBy: root },
+      { action: childEdit, invalidatedBy: root },
+    ]);
+    expect(retry.invalidatedByVeto).toEqual(first.invalidatedByVeto);
+    // Vetoed and invalidated records are deleted so read overlays recompute without them.
+    expect(store.getAction(root)).toBeUndefined();
+    expect(store.getAction(childEdit)).toBeUndefined();
+    expect(store.knowsProvisional("~root")).toBe(false);
+    expect(calls.addComment).toHaveLength(0);
+  });
+
+  it("makes the legacy single-action path throw for a cascade-invalidated action", async () => {
+    const { api } = makeApi();
+    const store = storeWith(api);
+    const root = stage(store, {
+      type: "createContent", provisionalId: "~root", kind: "page",
+      parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current",
+    });
+    const edit = stage(store, {
+      type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root",
+    });
+
+    await applyStoredActionsThrough(store, root, [root]);
+
+    // An un-migrated overseer applying the orphan must see a failure, not a silent success.
+    await expect(applyStoredAction(store, edit)).rejects.toThrow(`Unknown action: ${edit}`);
+  });
+
+  it("ignores a veto of an already-applied action", async () => {
+    const { api } = makeApi();
+    const store = storeWith(api);
+    const id = stage(store, { type: "addComment", contentId: "page", text: "Comment" });
+    await applyStoredActionsThrough(store, id, []);
+
+    const result = await applyStoredActionsThrough(store, id, [id]);
+
+    expect(result).toEqual({});
+    expect(store.getAction(id)?.state).toBe("applied");
+  });
+
+  it("rejects an out-of-range veto before changing state", async () => {
+    const { api, calls } = makeApi();
+    const store = storeWith(api);
+    const id = stage(store, { type: "addComment", contentId: "page", text: "Comment" });
+
+    await expect(applyStoredActionsThrough(store, id, [id + 1]))
+      .rejects.toThrow("Invalid veto action ID");
+
+    expect(store.getAction(id)?.state).toBe("pending");
+    expect(calls.addComment).toHaveLength(0);
+  });
+
+  it("persists later vetoes before an earlier action fails", async () => {
+    const { api } = makeApi();
+    const store = storeWith(api);
+    const failed = stage(store, { type: "addComment", contentId: "failed", text: "Comment" });
+    const vetoed = stage(store, {
+      type: "createContent", provisionalId: "~root", kind: "page",
+      parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current",
+    });
+    const invalidated = stage(store, {
+      type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root",
+    });
+    api.addComment = async () => { throw new Error("safe failure"); };
+
+    const result = await applyStoredActionsThrough(store, invalidated, [vetoed]);
+
+    expect(result.stopped?.at).toBe(failed);
+    expect(result.invalidatedByVeto).toEqual([{ action: invalidated, invalidatedBy: vetoed }]);
+    expect(store.getAction(vetoed)).toBeUndefined();
+    expect(store.getAction(invalidated)).toBeUndefined();
+  });
+});
+
+describe("stageAction", () => {
+  it("keeps the record staged until submitAction completes", async () => {
+    const { api } = makeApi();
+    const store = storeWith(api);
+    let stateDuringSubmit: string | undefined;
+    const approvalQueue = {
+      submitAction: async (id: number) => {
+        stateDuringSubmit = store.getAction(id)?.state;
+      },
+    } as unknown as Parameters<typeof stageAction>[1];
+
+    const id = await stageAction(store, approvalQueue, {
+      type: "addComment", contentId: "page", text: "Comment",
+    });
+
+    expect(stateDuringSubmit).toBe("staged");
+    expect(store.getAction(id)?.state).toBe("pending");
+  });
+
+  it("rolls the record back when submitAction fails", async () => {
+    const { api } = makeApi();
+    const store = storeWith(api);
+    const approvalQueue = {
+      submitAction: async () => { throw new Error("submit failed"); },
+    } as unknown as Parameters<typeof stageAction>[1];
+
+    await expect(stageAction(store, approvalQueue, {
+      type: "addComment", contentId: "page", text: "Comment",
+    })).rejects.toThrow("submit failed");
+
+    expect(store.allActions()).toHaveLength(0);
+  });
+});
+
 describe("revertStoredAction", () => {
   it("marks the action reverted on success", async () => {
     const { api } = makeApi();