diff --git a/packages/backend-utils/__tests__/gatekeeper-action.test.ts b/packages/backend-utils/__tests__/gatekeeper-action.test.ts new file mode 100644 index 000000000..92bbf6e30 --- /dev/null +++ b/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[0]; + +// Minimal in-memory KV matching the slice of the DO storage API the log uses. +function makeKv(): Kv { + const map = new Map(); + return { + put: (k: string, v: unknown) => void map.set(k, v), + delete: (k: string) => void map.delete(k), + list: ({ 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)["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(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); + } + }); +}); diff --git a/packages/backend-utils/package.json b/packages/backend-utils/package.json index db0da112d..76646be8d 100644 --- a/packages/backend-utils/package.json +++ b/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": { diff --git a/packages/backend-utils/src/gatekeeper-action.ts b/packages/backend-utils/src/gatekeeper-action.ts new file mode 100644 index 000000000..45fc3ed36 --- /dev/null +++ b/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 = Promise.resolve(); + + /** Enqueues an operation without allowing a rejection to block later operations. */ + run(operation: () => T | Promise): Promise { + 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 { + if (!Number.isSafeInteger(actionId) || actionId < 1) throw new TypeError("Invalid action ID."); + const result = new Set(); + 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): VetoInvalidation[] { + return [...this.#kv.list({ 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, pendingFloor: number): void { + for (const [key, vetoedBy] of this.#kv.list({ 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)}`); +} diff --git a/packages/backend-utils/tsconfig.json b/packages/backend-utils/tsconfig.json index ae8d4640b..3b380fab4 100644 --- a/packages/backend-utils/tsconfig.json +++ b/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"] diff --git a/packages/gatekeeper-confluence/__tests__/apply.test.ts b/packages/gatekeeper-confluence/__tests__/apply.test.ts index 7b0385dac..20566bd77 100644 --- a/packages/gatekeeper-confluence/__tests__/apply.test.ts +++ b/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,208 @@ 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, []); + + // The staged action can't be applied but must not be silently skipped either (the caller + // would infer everything through `pending` was applied), so the pass stops at it. + expect(result.stopped).toMatchObject({ at: staged, reason: expect.any(Error) }); + expect(calls.addComment).toHaveLength(0); + expect(store.getAction(staged)?.state).toBe("staged"); + expect(store.getAction(pending)?.state).toBe("pending"); + }); + + 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[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[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(); diff --git a/packages/gatekeeper-confluence/package.json b/packages/gatekeeper-confluence/package.json index 9bb0bbb41..015e7eb9e 100644 --- a/packages/gatekeeper-confluence/package.json +++ b/packages/gatekeeper-confluence/package.json @@ -10,6 +10,7 @@ "test:run": "vitest run" }, "dependencies": { + "@gadgets/backend-utils": "workspace:*", "@gadgets/configurator-ui": "workspace:*", "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", diff --git a/packages/gatekeeper-confluence/src/confluence-actions.ts b/packages/gatekeeper-confluence/src/confluence-actions.ts index db6dc8953..c8ffbcd20 100644 --- a/packages/gatekeeper-confluence/src/confluence-actions.ts +++ b/packages/gatekeeper-confluence/src/confluence-actions.ts @@ -9,7 +9,18 @@ // without it. import type { RpcStub } from "cloudflare:workers"; -import type { ActionDescription, ApprovalQueue, ObservationDescription } from "@gadgets/workshop-shared/gatekeeper"; +import { + InvalidationLog, + displayReason, + validateApplyThroughArgs, +} from "@gadgets/backend-utils/gatekeeper-action"; +import { createLogger } from "@gadgets/backend-utils/logger"; +import type { + ActionDescription, + ApplyActionsThroughResult, + ApprovalQueue, + ObservationDescription, +} from "@gadgets/workshop-shared/gatekeeper"; import { ConfluenceApi, contentBodyMarkdown, @@ -19,6 +30,15 @@ import { import { markdownToStorage, storageToMarkdown } from "./confluence-markdown"; import type { Comment, ContentSummary, ContentType } from "./types"; +/** Observability fields emitted by Confluence action resolution. */ +type ConfluenceActionLogFields = { actionId: number; vendorId: string }; + +const VENDOR_ID = "confluence"; + +const logger = createLogger({ + component: "gatekeeper.confluence.actions", vendorId: VENDOR_ID, +}); + // --------------------------------------------------------------------------------------------- // Action model @@ -56,7 +76,11 @@ export type ConfluenceAction = export type StoredActionRecord = { id: number; action: ConfluenceAction; - state: "pending" | "applied" | "reverted"; + /** + * "staged" means submitAction() has not completed yet: the record overlays reads like a pending + * one, but applyStoredActionsThrough() must not apply it. + */ + state: "staged" | "pending" | "applied" | "reverted"; submittedAt: number; /** For createContent / addComment: the real content ID assigned on apply. */ createdContentId?: string; @@ -81,9 +105,13 @@ export class ConfluenceStore { #kv: Kv; #api: ConfluenceApi; + /** Durable attribution of veto-cascade invalidations, re-reported on repeated requests. */ + readonly invalidations: InvalidationLog; + constructor(kv: Kv, api: ConfluenceApi) { this.#kv = kv; this.#api = api; + this.invalidations = new InvalidationLog(kv); } get api(): ConfluenceApi { @@ -124,8 +152,9 @@ export class ConfluenceStore { .toSorted((a, b) => a.id - b.id); } + /** Not-yet-applied actions, including staged ones (read overlays must reflect both). */ pendingActions(): StoredActionRecord[] { - return this.allActions().filter(r => r.state === "pending"); + return this.allActions().filter(r => r.state === "pending" || r.state === "staged"); } /** Pending actions targeting a piece of content (addressed by either provisional or real ID). */ @@ -417,18 +446,25 @@ function truncate(text: string, max = 2000): string { // --------------------------------------------------------------------------------------------- // Staging -/** Record a pending action and submit it for approval. Rolls back the record if submit fails. */ +/** Record a staged action and submit it for approval. Rolls back the record if submit fails. */ export async function stageAction( store: ConfluenceStore, approvalQueue: RpcStub, action: ConfluenceAction, ): Promise { const id = store.nextActionId(); - store.putAction({ id, action, state: "pending", submittedAt: Date.now() }); + store.putAction({ id, action, state: "staged", submittedAt: Date.now() }); try { await approvalQueue.submitAction(id, describeAction(action)); } catch (err) { store.deleteAction(id); throw err; } + // Only now may the action be applied: the overseer has accepted it, so a decision frontier can + // legitimately cover it. A concurrent veto cascade may have deleted the record meanwhile. + const record = store.getAction(id); + if (record?.state === "staged") { + record.state = "pending"; + store.putAction(record); + } return id; } @@ -568,43 +604,100 @@ export async function applyStoredAction(store: ConfluenceStore, id: number): Pro await applyAction(store, record); } -export function rejectStoredAction(store: ConfluenceStore, id: number): void | { restart?: boolean } { - const record = store.getAction(id); - if (!record) return; - store.deleteAction(id); - - // Rejecting a creation invalidates any pending actions on that (now-nonexistent) content, - // including child pages created under it, transitively. Cascade-delete and request a restart. - if (record.action.type === "createContent") { - const pending = store.pendingActions(); - const purge = new Set([record.action.provisionalId]); - for (;;) { - let added = false; - for (const r of pending) { - if (r.action.type === "createContent" && r.action.parent.type === "page" && - purge.has(r.action.parent.parentId) && !purge.has(r.action.provisionalId)) { - purge.add(r.action.provisionalId); - added = true; - } +/** + * Rejecting a creation invalidates any pending actions on that (now-nonexistent) content, + * including child pages created under it, transitively. Cascade-delete them all, recording which + * veto invalidated each so a repeated request can re-report the attribution. + */ +function cascadeRejectedCreation(store: ConfluenceStore, record: StoredActionRecord): void { + if (record.action.type !== "createContent") return; + + // Snapshot the pending set once — deleting actions below would otherwise change it under us. + const pending = store.pendingActions(); + const purge = new Set([record.action.provisionalId]); + for (;;) { + let added = false; + for (const candidate of pending) { + if (candidate.action.type === "createContent" && candidate.action.parent.type === "page" && + purge.has(candidate.action.parent.parentId) && !purge.has(candidate.action.provisionalId)) { + purge.add(candidate.action.provisionalId); + added = true; } - if (!added) break; } - let deleted = false; - for (const r of pending) { - const t = actionContentId(r.action); - if (t !== null && purge.has(t)) { - store.deleteAction(r.id); - deleted = true; - } + if (!added) break; + } + + for (const candidate of pending) { + const target = actionContentId(candidate.action); + if (target !== null && purge.has(target)) { + store.deleteAction(candidate.id); + store.invalidations.record(candidate.id, record.id); + } + } +} + +/** Delete vetoed records and cascade to actions they invalidate. Settled records are left alone. */ +function rejectRecords(store: ConfluenceStore, vetoes: Set): void { + const rejected: StoredActionRecord[] = []; + for (const id of vetoes) { + const record = store.getAction(id); + if (!record || record.state === "applied" || record.state === "reverted") continue; + store.deleteAction(id); + rejected.push(record); + } + for (const record of rejected) cascadeRejectedCreation(store, record); +} + +/** Resolve all stored actions through a Gatekeeper-local action ID. */ +export async function applyStoredActionsThrough( + store: ConfluenceStore, actionId: number, vetoes: number[], +): Promise { + const vetoSet = validateApplyThroughArgs(actionId, vetoes); + rejectRecords(store, vetoSet); + store.invalidations.prune(vetoSet, store.pendingActions()[0]?.id ?? Infinity); + + const invalidatedByVeto = store.invalidations.attributedTo(vetoSet); + const invalidations = invalidatedByVeto.length > 0 ? { invalidatedByVeto } : {}; + for (const record of store.pendingActions()) { + if (record.id > actionId) break; + if (record.state === "staged") { + // submitAction() has not completed, so this action must not be applied yet -- and the + // contract forbids silently skipping an in-range action (the caller would infer it was + // applied), so report it as the stopping point. The submit completes momentarily and the + // next pass proceeds. + return { + ...invalidations, + stopped: { + at: record.id, + reason: new Error("This action is still being submitted for approval. Retry in a moment."), + }, + }; + } + try { + await applyAction(store, record); + } catch (error) { + logger.warn("failed to apply action", { + event: "action.apply.failed", + actionId: record.id, + error, + }); + return { + ...invalidations, + stopped: { + at: record.id, + reason: displayReason(error, "Confluence could not apply this action"), + }, + }; } - return deleted ? { restart: true } : undefined; } + return invalidations; +} - const target = actionContentId(record.action); - if (target && store.pendingForContent(target).length > 0) return { restart: true }; +export function rejectStoredAction(store: ConfluenceStore, id: number): void { + rejectRecords(store, new Set([id])); } -type RevertResult = void | { message?: string; canRetry?: boolean; restart?: boolean }; +type RevertResult = void | { message?: string; canRetry?: boolean }; export async function revertStoredAction(store: ConfluenceStore, id: number): Promise { const record = store.getAction(id); diff --git a/packages/gatekeeper-confluence/src/confluence.ts b/packages/gatekeeper-confluence/src/confluence.ts index a2a033f01..b98431ff4 100644 --- a/packages/gatekeeper-confluence/src/confluence.ts +++ b/packages/gatekeeper-confluence/src/confluence.ts @@ -15,6 +15,7 @@ // 7. Observer verification — all bindings track independently restricted spaces and content. import { DurableObject, RpcStub, RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; +import { SerialTaskQueue } from "@gadgets/backend-utils/gatekeeper-action"; import { skipRpcValidation, validateRpc } from "capnweb-validate"; import { type AccountDescription, @@ -58,6 +59,7 @@ import { } from "./confluence-api"; import { ConfluenceStore, + applyStoredActionsThrough, applyStoredAction, observation, overlayChildPages, @@ -568,6 +570,7 @@ function makeApi(ctx: { exports: Cloudflare.Env }, props: BaseProps): Confluence @validateRpc() export class ConfluenceSiteGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -609,14 +612,24 @@ export class ConfluenceSiteGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } @validateRpc() export class ConfluenceSpaceGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -655,14 +668,24 @@ export class ConfluenceSpaceGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } @validateRpc() export class ConfluenceContentGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -702,9 +725,18 @@ export class ConfluenceContentGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } function accountFor(ctx: { exports: Cloudflare.Env }, userObjectId: string) { diff --git a/packages/gatekeeper-notion/__tests__/notion-actions.test.ts b/packages/gatekeeper-notion/__tests__/notion-actions.test.ts new file mode 100644 index 000000000..78316c6a1 --- /dev/null +++ b/packages/gatekeeper-notion/__tests__/notion-actions.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vitest"; +import { + NotionStore, + applyStoredActionsThrough, + type NotionAction, +} from "../src/notion-actions"; +import type { NotionApi } from "../src/notion-api"; + +type Kv = ConstructorParameters[0]; + +function makeKv(): Kv { + const map = new Map(); + return { + get: (key: string) => map.get(key) as T | undefined, + put: (key: string, value: unknown) => void map.set(key, value), + delete: (key: string) => void map.delete(key), + list: ({ prefix }: { prefix: string }) => + [...map.entries()].filter(([key]) => key.startsWith(prefix)) as [string, T][], + } as unknown as Kv; +} + +function makeStore() { + const comments: string[] = []; + const api = { + createComment: async ({ parent }: { parent: { page_id: string } }) => { + comments.push(parent.page_id); + if (parent.page_id === "failed") throw new Error("safe failure"); + return {}; + }, + } as unknown as NotionApi; + return { store: new NotionStore(makeKv(), api), comments }; +} + +function stage(store: NotionStore, action: NotionAction): number { + const id = store.nextActionId(); + store.putAction({ id, action, state: "pending", submittedAt: id }); + return id; +} + +describe("applyStoredActionsThrough", () => { + it("skips sparse IDs and stops at the first failed action", async () => { + const { store, comments } = makeStore(); + const first = stage(store, { type: "addComment", pageId: "first", text: "one" }); + const hole = stage(store, { type: "addComment", pageId: "hole", text: "two" }); + const failed = stage(store, { type: "addComment", pageId: "failed", text: "three" }); + const later = stage(store, { type: "addComment", pageId: "later", text: "four" }); + store.deleteAction(hole); + + const result = await applyStoredActionsThrough(store, later, []); + + expect(comments).toEqual(["first", "failed"]); + expect(store.getAction(first)?.state).toBe("applied"); + expect(store.getAction(failed)?.state).toBe("pending"); + expect(store.getAction(later)?.state).toBe("pending"); + expect(result.stopped).toMatchObject({ at: failed, reason: expect.any(Error) }); + }); + + it("stops at a staged action whose submission has not completed", async () => { + const { store, comments } = makeStore(); + const staged = store.nextActionId(); + store.putAction({ + id: staged, action: { type: "addComment", pageId: "staged", text: "early" }, + state: "staged", submittedAt: staged, + }); + const pending = stage(store, { type: "addComment", pageId: "later", text: "late" }); + + const result = await applyStoredActionsThrough(store, pending, []); + + // The staged action can't be applied but must not be silently skipped either (the caller + // would infer everything through `pending` was applied), so the pass stops at it. + expect(result.stopped).toMatchObject({ at: staged, reason: expect.any(Error) }); + expect(comments).toHaveLength(0); + expect(store.getAction(staged)?.state).toBe("staged"); + expect(store.getAction(pending)?.state).toBe("pending"); + }); + + it("persists transitive invalidations and reports them again on retry", async () => { + const { store, comments } = makeStore(); + const root = stage(store, { + type: "createPage", provisionalId: "~root", parent: { kind: "workspace" }, title: "Root", + }); + const child = stage(store, { + type: "createPage", provisionalId: "~child", parent: { kind: "page", pageId: "~root" }, + title: "Child", + }); + const edit = stage(store, { type: "addComment", pageId: "~child", text: "Comment" }); + + const first = await applyStoredActionsThrough(store, root, [root]); + const retry = await applyStoredActionsThrough(store, root, [root]); + + expect(first.invalidatedByVeto).toEqual([ + { action: child, invalidatedBy: root }, + { action: edit, 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(edit)).toBeUndefined(); + expect(store.knowsProvisional("~root")).toBe(false); + expect(comments).toHaveLength(0); + }); + + it("ignores a veto of an already-applied action", async () => { + const { store, comments } = makeStore(); + const id = stage(store, { type: "addComment", pageId: "page", text: "Comment" }); + await applyStoredActionsThrough(store, id, []); + + const result = await applyStoredActionsThrough(store, id, [id]); + + expect(result).toEqual({}); + expect(store.getAction(id)?.state).toBe("applied"); + expect(comments).toEqual(["page"]); + }); + + it("rejects an out-of-range veto before changing state", async () => { + const { store, comments } = makeStore(); + const id = stage(store, { type: "addComment", pageId: "page", text: "Comment" }); + + await expect(applyStoredActionsThrough(store, id, [id + 1])) + .rejects.toThrow("Invalid veto action ID"); + + expect(store.getAction(id)?.state).toBe("pending"); + expect(comments).toHaveLength(0); + }); +}); diff --git a/packages/gatekeeper-notion/package.json b/packages/gatekeeper-notion/package.json index 8dd05ac74..14f600534 100644 --- a/packages/gatekeeper-notion/package.json +++ b/packages/gatekeeper-notion/package.json @@ -10,6 +10,7 @@ "test:run": "vitest run" }, "dependencies": { + "@gadgets/backend-utils": "workspace:*", "@gadgets/configurator-ui": "workspace:*", "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", diff --git a/packages/gatekeeper-notion/src/notion-actions.ts b/packages/gatekeeper-notion/src/notion-actions.ts index a43248503..a638f245c 100644 --- a/packages/gatekeeper-notion/src/notion-actions.ts +++ b/packages/gatekeeper-notion/src/notion-actions.ts @@ -27,7 +27,18 @@ import { type NotionPageResponse, } from "./notion-api"; import type { RpcStub } from "cloudflare:workers"; -import type { ActionDescription, ApprovalQueue, ObservationDescription } from "@gadgets/workshop-shared/gatekeeper"; +import { + InvalidationLog, + displayReason, + validateApplyThroughArgs, +} from "@gadgets/backend-utils/gatekeeper-action"; +import { createLogger } from "@gadgets/backend-utils/logger"; +import type { + ActionDescription, + ApplyActionsThroughResult, + ApprovalQueue, + ObservationDescription, +} from "@gadgets/workshop-shared/gatekeeper"; import type { NotionComment, NotionDatabaseSchema, @@ -40,6 +51,15 @@ import type { NotionUser, } from "./types"; +/** Observability fields emitted by Notion action resolution. */ +type NotionActionLogFields = { actionId: number; vendorId: string }; + +const VENDOR_ID = "notion"; + +const logger = createLogger({ + component: "gatekeeper.notion.actions", vendorId: VENDOR_ID, +}); + // --------------------------------------------------------------------------------------------- // Action model @@ -77,7 +97,11 @@ export type NotionAction = export type StoredActionRecord = { id: number; action: NotionAction; - state: "pending" | "applied" | "reverted"; + /** + * "staged" means submitAction() has not completed yet: the record overlays reads like a pending + * one, but applyStoredActionsThrough() must not apply it. + */ + state: "staged" | "pending" | "applied" | "reverted"; submittedAt: number; /** For appendContent revert: the IDs of the blocks created on apply. */ appendedBlockIds?: string[]; @@ -110,9 +134,13 @@ export class NotionStore { #kv: Kv; #api: NotionApi; + /** Durable attribution of veto-cascade invalidations, re-reported on repeated requests. */ + readonly invalidations: InvalidationLog; + constructor(kv: Kv, api: NotionApi) { this.#kv = kv; this.#api = api; + this.invalidations = new InvalidationLog(kv); } get api(): NotionApi { @@ -153,8 +181,9 @@ export class NotionStore { .toSorted((a, b) => a.id - b.id); } + /** Not-yet-applied actions, including staged ones (read overlays must reflect both). */ pendingActions(): StoredActionRecord[] { - return this.allActions().filter(r => r.state === "pending"); + return this.allActions().filter(r => r.state === "pending" || r.state === "staged"); } /** @@ -820,7 +849,7 @@ export async function applyNotionAction(store: NotionStore, record: StoredAction export async function revertNotionAction( store: NotionStore, record: StoredActionRecord, -): Promise { +): Promise { const api = store.api; const action = record.action; @@ -1004,7 +1033,7 @@ export function buildCreateBody( } /** - * Record a pending action and submit it to the approval queue for later approval. If the submit + * Record a staged action and submit it to the approval queue for later approval. If the submit * fails, the stored record is rolled back so it doesn't pollute simulation. Returns the action ID. */ export async function stageAction( @@ -1013,13 +1042,20 @@ export async function stageAction( action: NotionAction, ): Promise { const id = store.nextActionId(); - store.putAction({ id, action, state: "pending", submittedAt: Date.now() }); + store.putAction({ id, action, state: "staged", submittedAt: Date.now() }); try { await approvalQueue.submitAction(id, describeAction(action)); } catch (err) { store.deleteAction(id); throw err; } + // Only now may the action be applied: the overseer has accepted it, so a decision frontier can + // legitimately cover it. A concurrent veto cascade may have deleted the record meanwhile. + const record = store.getAction(id); + if (record?.state === "staged") { + record.state = "pending"; + store.putAction(record); + } return id; } @@ -1032,57 +1068,107 @@ export async function applyStoredAction(store: NotionStore, id: number): Promise await applyNotionAction(store, record); } -export function rejectStoredAction(store: NotionStore, id: number): void | { restart?: boolean } { - const record = store.getAction(id); - if (!record) return; - store.deleteAction(id); - - // Rejecting a page creation invalidates every pending action that targeted that provisional page - // (they could never be applied — the page won't exist), including sub-pages created under it and - // their edits, transitively. Cascade-delete them all so the overseer never tries to apply an - // orphan, and request a restart since the Gadget already observed simulated state built on them. - if (record.action.type === "createPage") { - // Snapshot the pending set once — deleting actions below would otherwise change it under us. - const pending = store.pendingActions(); - const purge = new Set([record.action.provisionalId]); - // Expand to transitively-nested sub-page creations. Only `createSubPage` nests a creation under - // a provisional page (parent.kind === "page"); database/workspace creates never have a - // provisional parent, so they don't need handling here. - for (;;) { - let added = false; - for (const r of pending) { - if (r.action.type === "createPage" && r.action.parent.kind === "page" && - purge.has(r.action.parent.pageId) && !purge.has(r.action.provisionalId)) { - purge.add(r.action.provisionalId); - added = true; - } +/** + * Rejecting a page creation invalidates every pending action that targeted that provisional page + * (they could never be applied — the page won't exist), including sub-pages created under it and + * their edits, transitively. Cascade-delete them all, recording which veto invalidated each so a + * repeated request can re-report the attribution. + */ +function cascadeRejectedCreation(store: NotionStore, record: StoredActionRecord): void { + if (record.action.type !== "createPage") return; + + // Snapshot the pending set once — deleting actions below would otherwise change it under us. + const pending = store.pendingActions(); + const purge = new Set([record.action.provisionalId]); + // Expand to transitively-nested sub-page creations. Only `createSubPage` nests a creation under + // a provisional page (parent.kind === "page"); database/workspace creates never have a + // provisional parent, so they don't need handling here. + for (;;) { + let added = false; + for (const candidate of pending) { + if (candidate.action.type === "createPage" && candidate.action.parent.kind === "page" && + purge.has(candidate.action.parent.pageId) && !purge.has(candidate.action.provisionalId)) { + purge.add(candidate.action.provisionalId); + added = true; } - if (!added) break; } - let deleted = false; - for (const r of pending) { - const t = actionPageId(r.action); - if (t !== null && purge.has(t)) { - store.deleteAction(r.id); - deleted = true; - } + if (!added) break; + } + + for (const candidate of pending) { + const target = actionPageId(candidate.action); + if (target !== null && purge.has(target)) { + store.deleteAction(candidate.id); + store.invalidations.record(candidate.id, record.id); } - if (deleted) return { restart: true }; - return; } +} + +/** Delete vetoed records and cascade to actions they invalidate. Settled records are left alone. */ +function rejectRecords(store: NotionStore, vetoes: Set): void { + const rejected: StoredActionRecord[] = []; + for (const id of vetoes) { + const record = store.getAction(id); + if (!record || record.state === "applied" || record.state === "reverted") continue; + store.deleteAction(id); + rejected.push(record); + } + for (const record of rejected) cascadeRejectedCreation(store, record); +} - // Rejecting a mid-stack edit leaves the simulated overlay the Gadget already observed - // inconsistent; ask for a restart if other pending actions still target the same page. - const target = actionPageId(record.action); - if (target && store.pendingForPage(target).length > 0) { - return { restart: true }; +/** Resolve all stored actions through a Gatekeeper-local action ID. */ +export async function applyStoredActionsThrough( + store: NotionStore, actionId: number, vetoes: number[], +): Promise { + const vetoSet = validateApplyThroughArgs(actionId, vetoes); + rejectRecords(store, vetoSet); + store.invalidations.prune(vetoSet, store.pendingActions()[0]?.id ?? Infinity); + + const invalidatedByVeto = store.invalidations.attributedTo(vetoSet); + const invalidations = invalidatedByVeto.length > 0 ? { invalidatedByVeto } : {}; + for (const record of store.pendingActions()) { + if (record.id > actionId) break; + if (record.state === "staged") { + // submitAction() has not completed, so this action must not be applied yet -- and the + // contract forbids silently skipping an in-range action (the caller would infer it was + // applied), so report it as the stopping point. The submit completes momentarily and the + // next pass proceeds. + return { + ...invalidations, + stopped: { + at: record.id, + reason: new Error("This action is still being submitted for approval. Retry in a moment."), + }, + }; + } + try { + await applyNotionAction(store, record); + } catch (error) { + logger.warn("failed to apply action", { + event: "action.apply.failed", + actionId: record.id, + error, + }); + return { + ...invalidations, + stopped: { + at: record.id, + reason: displayReason(error, "Notion could not apply this action"), + }, + }; + } } + return invalidations; +} + +export function rejectStoredAction(store: NotionStore, id: number): void { + rejectRecords(store, new Set([id])); } export async function revertStoredAction( store: NotionStore, id: number, -): Promise { +): Promise { const record = store.getAction(id); if (!record) throw new Error(`Unknown action: ${id}`); return await revertNotionAction(store, record); diff --git a/packages/gatekeeper-notion/src/notion.ts b/packages/gatekeeper-notion/src/notion.ts index af56af206..ed6f32f53 100644 --- a/packages/gatekeeper-notion/src/notion.ts +++ b/packages/gatekeeper-notion/src/notion.ts @@ -14,6 +14,7 @@ // writes immediately. List simulation has documented limitations (see types.d.ts). import { DurableObject, RpcStub, RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; +import { SerialTaskQueue } from "@gadgets/backend-utils/gatekeeper-action"; import { skipRpcValidation, validateRpc } from "capnweb-validate"; import { stripTrailingSlashes, @@ -48,6 +49,7 @@ import { } from "./notion-api"; import { NotionStore, + applyStoredActionsThrough, applyStoredAction, defaultPropertiesFromSchema, observation, @@ -694,6 +696,7 @@ type NotionItemGatekeeperImplProps = { @validateRpc() export class NotionItemGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #api(): NotionApi { const userObjectId = this.ctx.props.userObjectId; const account = () => @@ -778,16 +781,20 @@ export class NotionItemGatekeeperImpl extends DurableObject {} - async applyAction(action: number): Promise { - await applyStoredAction(this.#store(), action); + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); } - async rejectAction(action: number): Promise { - return rejectStoredAction(this.#store(), action); + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); } - async revertAction(action: number) { - return await revertStoredAction(this.#store(), action); + rejectAction(action: number): Promise { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); } } @@ -799,6 +806,7 @@ type NotionWorkspaceGatekeeperImplProps = { export class NotionWorkspaceGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #api(): NotionApi { const userObjectId = this.ctx.props.userObjectId; const account = () => @@ -940,16 +948,20 @@ export class NotionWorkspaceGatekeeperImpl this.ctx.storage.kv.delete(this.#observerKey(id)); } - async applyAction(action: number): Promise { - await applyStoredAction(this.#store(), action); + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); } - async rejectAction(action: number): Promise { - return rejectStoredAction(this.#store(), action); + rejectAction(action: number): Promise { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); } - async revertAction(action: number) { - return await revertStoredAction(this.#store(), action); + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); } } diff --git a/packages/workshop-shared/src/gatekeeper.ts b/packages/workshop-shared/src/gatekeeper.ts index d642dbf56..7eb22e433 100644 --- a/packages/workshop-shared/src/gatekeeper.ts +++ b/packages/workshop-shared/src/gatekeeper.ts @@ -684,6 +684,35 @@ export interface GatekeeperUser extends WorkerEntrypoint { */ export interface GatekeeperUserVerifier extends WorkerEntrypoint {} +/** Result of applying a Gatekeeper's queued actions through a decision frontier. */ +export interface ApplyActionsThroughResult { + /** + * Something went unexpectedly wrong at the given action number; remaining actions were not + * applied. The user may retry after resolving the problem or vetoing. + */ + stopped?: { + /** Gatekeeper-local action ID that could not be applied. */ + at: number; + + /** + * Explanation of why application stopped. Only the error's `message` survives the RPC hop, + * so it must stand alone as display-safe text, specific enough for the user to resolve the + * problem. + */ + reason: Error; + }; + + /** + * Actions invalidated as a result of vetoes (may include future action numbers). Display-only. + * Gatekeepers MAY not track dependencies and instead let the dependent action fail with + * `stopped`. + * + * Each entry pairs an invalidated Gatekeeper-local action ID with the vetoed action ID that + * invalidated it. + */ + invalidatedByVeto?: Array<{action: number, invalidatedBy: number}>; +} + /** * Interface exposed by a Gatekeeper instance implementing a specific resource binding on a * specific Gadget. @@ -794,33 +823,39 @@ export interface Gatekeeper extends DurableObject { getSlashCommandProvider?(): Promise; // --------------------------------------------------------------------------- - // Callbacks invoked by the overseer to apply (or reject) actions that were previously queued - // for approval via the ApprovalQueue. + // Callback invoked by the overseer to resolve actions that were previously queued for approval + // via the ApprovalQueue. // // Each action is identified by a sequential integer action ID, assigned by the gatekeeper when - // it submits the action for approval. The action ID is passed back to these methods so the + // it submits the action for approval. The action ID is passed back to this method so the // gatekeeper can look up the action details in its own storage. /** - * Action was approved. This call should apply the action (or schedule it to be applied). + * Applies all actions through the given action ID (includes all previous actions that are not + * yet applied). Action IDs listed in `vetoes` are actions the user has rejected. + * + * Actions are applied in ascending ID order. Vetoed actions and actions invalidated by a veto + * become terminal no-ops. Processing stops at the first application failure; a pending in-range + * action the gatekeeper still holds must never be silently skipped — it is either applied or + * reported via `stopped`. An action whose `submitAction()` call has not yet completed must not + * be applied. * - * If this throws an exception, the user will be informed that the action failed and given the - * opportunity to retry or discard. + * `actionId` is the decision frontier and may equal the current frontier to deliver vetoes only. + * Every action ID in `vetoes` must be less than or equal to `actionId`. A veto may arrive long + * after the user rejected the action; delivery is opportunistic, not prompt. * - * Depending on policy conditions, an action may be approved and applied automatically. However, - * the gatekeeper is nevertheless expected to submit all actions for approval; there is no mode - * in which it's OK to skip the check. + * Calls must be idempotent. Missing IDs and vetoes of unknown or already-applied actions are + * ignored. A repeated request must re-report persisted invalidations attributable to its vetoes. */ + applyActionsThrough?(actionId: number, vetoes: number[]): Promise; + + /** @deprecated Implement `applyActionsThrough()` instead. */ applyAction(action: number): Promise; /** - * Indicates that an action was rejected by the user. The gatekeeper should clean up any - * associated storage. + * The returned `restart` flag is ignored; the overseer discards it. * - * If the returned `restart` flag is true, rejecting this action requires restarting the Gadget. - * This is sometimes needed by gatekeepers that simulate actions as if they had been approved -- - * the session may be in a state that is difficult to roll back without confusing the Gadget. - * The Overseer will take care of the restart, possibly after rejecting other actions. + * @deprecated Implement `applyActionsThrough()` instead. */ rejectAction(action: number): Promise; @@ -841,11 +876,9 @@ export interface Gatekeeper extends DurableObject { * `canRetry` should be true if the revert failed (for a reason described in `message`), but * it could make sense to retry later. In this case the UI will continue to give the user the * option to revert. - * - * `restart` has the same meaning as for `rejectAction()`. */ revertAction(action: number): - Promise; + Promise; } export interface ObservationAuthorizer extends RpcTarget { @@ -940,9 +973,8 @@ export interface ApprovalQueue extends ObservationAuthorizer { * be carried out until much later. It's intended that the user might not approve actions until * hours or days later, but this shouldn't cause any problems. * - * `action` is a sequential integer action ID assigned by the gatekeeper. It will be passed back - * to the Gatekeeper's applyAction() or rejectAction() when the action is later approved or - * rejected. + * `action` is a sequential integer action ID assigned by the gatekeeper. It will later be used as + * a decision frontier or veto in the Gatekeeper's `applyActionsThrough()` method. * * `description` describes the action in a way that can direct UI representation and policy * enforcement details. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f572acfe0..14fc76222 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -117,6 +117,9 @@ importers: packages/gatekeeper-confluence: dependencies: + '@gadgets/backend-utils': + specifier: workspace:* + version: link:../backend-utils '@gadgets/configurator-ui': specifier: workspace:* version: link:../configurator-ui @@ -465,6 +468,9 @@ importers: packages/gatekeeper-notion: dependencies: + '@gadgets/backend-utils': + specifier: workspace:* + version: link:../backend-utils '@gadgets/configurator-ui': specifier: workspace:* version: link:../configurator-ui @@ -1148,7 +1154,7 @@ packages: resolution: {integrity: sha512-ehg3jIkYKulZh+8om/O25vkvSsXXwC+skXmyA87FFx6A/45eqOkZsBltMw/TVteb0mloiGT8oGRTcjRAz66zaA==} '@cloudflare/kumo@2.9.2': - resolution: {integrity: sha512-c3RZBmx0TqxTKAPT4PWyTgVwPcDVW+KrFmf4mKCnwWBe6OIc0vWn+wMhnaARarJz/2kvsx87tMGmNRBsCn7pUA==} + resolution: {integrity: sha512-c3RZBmx0TqxTKAPT4PWyTgVwPcDVW+KrFmf4mKCnwWBe6OIc0vWn+wMhnaARarJz/2kvsx87tMGmNRBsCn7pUA==, tarball: https://registry.npmjs.org/@cloudflare/kumo/-/kumo-2.9.2.tgz} hasBin: true peerDependencies: '@phosphor-icons/react': ^2.1.10 @@ -1163,15 +1169,15 @@ packages: optional: true '@cloudflare/kv-asset-handler@0.5.0': - resolution: {integrity: sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==} + resolution: {integrity: sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==, tarball: https://registry.npmjs.org/@cloudflare/kv-asset-handler/-/kv-asset-handler-0.5.0.tgz} engines: {node: '>=22.0.0'} '@cloudflare/puppeteer@1.3.0': - resolution: {integrity: sha512-NBrJEUnqe082nopLh0eqnTXK4DjwsTsZGzoAcs71NFnBgzWU6Yb/ibUJHveCHV4AyAkM+mE/DChFev5gwaKZEg==} + resolution: {integrity: sha512-NBrJEUnqe082nopLh0eqnTXK4DjwsTsZGzoAcs71NFnBgzWU6Yb/ibUJHveCHV4AyAkM+mE/DChFev5gwaKZEg==, tarball: https://registry.npmjs.org/@cloudflare/puppeteer/-/puppeteer-1.3.0.tgz} engines: {node: '>=18'} '@cloudflare/unenv-preset@2.16.1': - resolution: {integrity: sha512-ECxObrMfyTl5bhQf/lZCXwo5G6xX9IAUo+nDMKK4SZ8m4Jvvxp52vilxyySSWh2YTZz8+HQ07qGH/2rEom1vDw==} + resolution: {integrity: sha512-ECxObrMfyTl5bhQf/lZCXwo5G6xX9IAUo+nDMKK4SZ8m4Jvvxp52vilxyySSWh2YTZz8+HQ07qGH/2rEom1vDw==, tarball: https://registry.npmjs.org/@cloudflare/unenv-preset/-/unenv-preset-2.16.1.tgz} peerDependencies: unenv: 2.0.0-rc.24 workerd: '>1.20260305.0 <2.0.0-0' @@ -1180,44 +1186,44 @@ packages: optional: true '@cloudflare/vitest-pool-workers@0.20.3': - resolution: {integrity: sha512-aCMvM5zQ3MTz8SorSZB6ZxjVK2Gof6UhIc2Z1z9zbX/obucSYwDUkx8A0MUOod4I7clfB0QLarKBjRdIczK3vg==} + resolution: {integrity: sha512-aCMvM5zQ3MTz8SorSZB6ZxjVK2Gof6UhIc2Z1z9zbX/obucSYwDUkx8A0MUOod4I7clfB0QLarKBjRdIczK3vg==, tarball: https://registry.npmjs.org/@cloudflare/vitest-pool-workers/-/vitest-pool-workers-0.20.3.tgz} peerDependencies: '@vitest/runner': ^4.1.0 '@vitest/snapshot': ^4.1.0 vitest: ^4.1.0 '@cloudflare/workerd-darwin-64@1.20260801.1': - resolution: {integrity: sha512-wuJWbXpKvncJi1P0GKS+iYpN5tHdb7JPJJ/+6ZQe8zzovHHVMkLJPNBsgWpqeUhpM3g9qTwEKd2rglNKejuh5A==} + resolution: {integrity: sha512-wuJWbXpKvncJi1P0GKS+iYpN5tHdb7JPJJ/+6ZQe8zzovHHVMkLJPNBsgWpqeUhpM3g9qTwEKd2rglNKejuh5A==, tarball: https://registry.npmjs.org/@cloudflare/workerd-darwin-64/-/workerd-darwin-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [darwin] '@cloudflare/workerd-darwin-arm64@1.20260801.1': - resolution: {integrity: sha512-kwoZiTpnhNrF3+APx84Q/oAqvJ3sU9yefGagwm/ASaH/2W19x0vghkW/r4qCoHCK0WW7EPugZ+aXjgPMRtlq1Q==} + resolution: {integrity: sha512-kwoZiTpnhNrF3+APx84Q/oAqvJ3sU9yefGagwm/ASaH/2W19x0vghkW/r4qCoHCK0WW7EPugZ+aXjgPMRtlq1Q==, tarball: https://registry.npmjs.org/@cloudflare/workerd-darwin-arm64/-/workerd-darwin-arm64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [arm64] os: [darwin] '@cloudflare/workerd-linux-64@1.20260801.1': - resolution: {integrity: sha512-r0vAxCZH+Jih9Unm1yoyiByPNWNgawcKciOHDm5Q37ZVGOkKLsT9AtLe3yLSaul76WrKqtf+JP2n0WW32VBLJg==} + resolution: {integrity: sha512-r0vAxCZH+Jih9Unm1yoyiByPNWNgawcKciOHDm5Q37ZVGOkKLsT9AtLe3yLSaul76WrKqtf+JP2n0WW32VBLJg==, tarball: https://registry.npmjs.org/@cloudflare/workerd-linux-64/-/workerd-linux-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [linux] '@cloudflare/workerd-linux-arm64@1.20260801.1': - resolution: {integrity: sha512-zWgpdZtSozvIgzQNmQiDSF8yEOQJUkRAWNsDXXzAAoy+fCn8YUoSibj3mpFSbZRvbUldeBEaW6SCdC2VEMkhNQ==} + resolution: {integrity: sha512-zWgpdZtSozvIgzQNmQiDSF8yEOQJUkRAWNsDXXzAAoy+fCn8YUoSibj3mpFSbZRvbUldeBEaW6SCdC2VEMkhNQ==, tarball: https://registry.npmjs.org/@cloudflare/workerd-linux-arm64/-/workerd-linux-arm64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [arm64] os: [linux] '@cloudflare/workerd-windows-64@1.20260801.1': - resolution: {integrity: sha512-2oQz+Ksu4ji6e/+ZoYX+tWQEcxAii2p7l+iR8kx48W1llMalaufAsmVxTlk+3/vrM7D3/2c0iK448e0UQTcIMg==} + resolution: {integrity: sha512-2oQz+Ksu4ji6e/+ZoYX+tWQEcxAii2p7l+iR8kx48W1llMalaufAsmVxTlk+3/vrM7D3/2c0iK448e0UQTcIMg==, tarball: https://registry.npmjs.org/@cloudflare/workerd-windows-64/-/workerd-windows-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [win32] '@cloudflare/workers-types@5.20260808.1': - resolution: {integrity: sha512-DN7G9SMyeOq031YhQexoExFAK78ms74cFiFF1teDlTK4+LjHgIc5Z8VbvXQtcCIP//o38btxzxW0b9MGPA83CA==} + resolution: {integrity: sha512-DN7G9SMyeOq031YhQexoExFAK78ms74cFiFF1teDlTK4+LjHgIc5Z8VbvXQtcCIP//o38btxzxW0b9MGPA83CA==, tarball: https://registry.npmjs.org/@cloudflare/workers-types/-/workers-types-5.20260808.1.tgz} '@codemirror/autocomplete@6.20.3': resolution: {integrity: sha512-tlosUqb+3BbxCxZdu4tKeRghPFC+QM7q4X5YhKV2eCmPG+1r2F3f4AaSz5sCrFqUtX4Jh20VFTKecl16MgiV9g==}