diff --git a/.changeset/approval-atomicity.md b/.changeset/approval-atomicity.md new file mode 100644 index 0000000000..2f7d2247ca --- /dev/null +++ b/.changeset/approval-atomicity.md @@ -0,0 +1,20 @@ +--- +"@executor-js/sdk": patch +--- + +fix: make pending-approval consumption atomic + +`PendingApprovalStore.consume` previously read the record and deleted it as +two separate operations. Two concurrent resumes (a double-click, a client +retry, or two hosts racing the same approval) could both read the record +before either deleted it, and both would execute the approved tool call — +duplicated side effects from a single approval. + +Consumption now goes through a new `BlobStore.compareAndDelete` primitive +with a single-winner guarantee: exactly one concurrent consumer observes the +record as present-and-removed; everyone else observes it as absent. The +in-memory store implements it as a synchronous Map operation (atomic in JS's +single-threaded model); the FumaDB-backed store implements it as +get+delete inside the serializing transaction the driver already provides +(libSQL/Postgres BEGIN/COMMIT). The approval's expiry and corrupt-record +semantics are unchanged. diff --git a/packages/core/sdk/src/approval-atomicity.test.ts b/packages/core/sdk/src/approval-atomicity.test.ts new file mode 100644 index 0000000000..f7cad493f3 --- /dev/null +++ b/packages/core/sdk/src/approval-atomicity.test.ts @@ -0,0 +1,128 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Predicate } from "effect"; + +import { makeInMemoryBlobStore, type BlobStore } from "./blob"; +import { + makePendingApprovalStore, + PENDING_APPROVAL_TTL_MS, + type PendingApproval, +} from "./pending-approval"; + +// --------------------------------------------------------------------------- +// Focused tests — pending-approval consume atomicity, deterministic layer. +// +// These tests pin the exactly-once consume invariant deterministically, +// plus the expiry/corrupt semantics required to survive the consume +// restructure. A companion fast-check property (concurrent consumes ⇒ +// exactly one non-null) lives elsewhere. +// +// NOTE on test infrastructure: +// - it.effect runs under @effect/vitest's TestContext scheduler, where +// Effect.sleep never advances — any async boundary deadlocks the test. +// Sync-only effects work; async ones must go through Effect.runPromise in +// a plain vitest test. +// - The concurrency proof needs the real scheduler AND an explicit +// read-barrier: without one, the synchronous Map store completes each +// consume before the next fiber starts, so the double-resume race can +// never manifest and the test would be vacuous as a concurrency proof. +// --------------------------------------------------------------------------- + +const approval = (overrides?: Partial): PendingApproval => ({ + executionId: "exec_1", + artifactId: "art_1", + code: 'return await tools.github.user.main.issues.create({"title":"x"})', + address: "github.user.main.issues.create", + expiresAt: Date.now() + PENDING_APPROVAL_TTL_MS, + ...overrides, +}); + +describe("approval consume atomicity (compareAndDelete gate)", () => { + it("exactly one of N concurrent consumes wins (single-winner invariant)", async () => { + const N = 8; + const backing = makeInMemoryBlobStore(); + // Promise-latch barrier: every fiber increments the arrival count after + // its read and parks on the shared promise, which resolves only when + // all N have arrived — every read happens before any delete. Pure JS + // promise semantics, immune to Effect scheduler/runtime differences. + let arrived = 0; + let releaseAll!: () => void; + const allArrived = new Promise((resolve) => { + releaseAll = resolve; + }); + const barrier: BlobStore = { + ...backing, + get: (ns, key) => + backing.get(ns, key).pipe( + Effect.tap(() => + Effect.sync(() => { + arrived++; + if (arrived === N) releaseAll(); + }), + ), + // tap (not andThen — that would discard the payload): park until + // every fiber has read, then pass the payload through untouched. + Effect.tap(() => Effect.promise(() => allArrived)), + ), + }; + + const outcome = await Effect.runPromise( + Effect.gen(function* () { + const store = makePendingApprovalStore(barrier, "u:t:s"); + yield* store.put(approval()); + + const results = yield* Effect.all( + Array.from({ length: N }, () => store.consume("exec_1")), + { concurrency: "unbounded" }, + ); + const winners = results.filter(Predicate.isNotNull); + + // Post-condition: the record is gone for everyone. + const after = yield* store.consume("exec_1"); + return { winners: winners.map((w) => w.address), after }; + }), + ); + + expect(outcome.winners.length).toBe(1); + expect(outcome.winners[0]).toBe("github.user.main.issues.create"); + expect(outcome.after).toBeNull(); + }); + + it.effect("a replayed consume after a win reads absent", () => + Effect.gen(function* () { + const store = makePendingApprovalStore(makeInMemoryBlobStore(), "u:t:s"); + yield* store.put(approval()); + expect(yield* store.consume("exec_1")).not.toBeNull(); + expect(yield* store.consume("exec_1")).toBeNull(); + expect(yield* store.consume("exec_1")).toBeNull(); + }), + ); + + it.effect("expired records are consumed-and-dropped (never retried)", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + let now = 1_000_000; + const store = makePendingApprovalStore(blobs, "u:t:s", () => now); + yield* store.put(approval({ expiresAt: now + 10 })); + + now += 100; // expires + expect(yield* store.consume("exec_1")).toBeNull(); + + // The record is gone — rolling the clock back does not resurrect it. + now = 1_000_000; + expect(yield* store.consume("exec_1")).toBeNull(); + }), + ); + + it.effect("corrupt records are consumed-and-dropped (never surfaced)", () => + Effect.gen(function* () { + const blobs = makeInMemoryBlobStore(); + yield* blobs.put("u:t:s/@pending-approval", "exec_1", "not-json{{"); + + const store = makePendingApprovalStore(blobs, "u:t:s"); + expect(yield* store.consume("exec_1")).toBeNull(); + + // Gone for good — a second consume finds nothing. + expect(yield* store.consume("exec_1")).toBeNull(); + }), + ); +}); diff --git a/packages/core/sdk/src/blob.ts b/packages/core/sdk/src/blob.ts index c98286611d..177651beca 100644 --- a/packages/core/sdk/src/blob.ts +++ b/packages/core/sdk/src/blob.ts @@ -38,6 +38,26 @@ export interface BlobStore { ) => Effect.Effect; readonly delete: (namespace: string, key: string) => Effect.Effect; readonly has: (namespace: string, key: string) => Effect.Effect; + /** + * Atomically delete a record IF it exists. Returns true iff this caller's + * delete removed an existing record; false iff it was already absent. + * + * The single-winner invariant: exactly one concurrent caller observes true + * for a given (namespace, key); everyone else observes false, and the + * post-condition is that the record is absent for all callers. There is no + * read-your-undefined window — a caller that observed true is guaranteed + * the record was present at deletion time, and no other caller can observe + * true for the same key afterwards. + * + * Implementations MUST NOT do check-then-act across two statements: + * either a single atomic statement with a rows-affected/count check, or + * get+delete inside a serializing transaction (libSQL/Postgres + * `fuma.transaction`), or a single synchronous Map op (in-memory). + */ + readonly compareAndDelete: ( + namespace: string, + key: string, + ) => Effect.Effect; } export interface PluginBlobStore { @@ -154,6 +174,16 @@ export const makeInMemoryBlobStore = (): BlobStore => { store.delete(k(ns, key)); }), has: (ns, key) => Effect.sync(() => store.has(k(ns, key))), + // Atomic by construction: a synchronous Map has+delete runs to completion + // without yielding, so no other fiber can interleave between the check + // and the delete in JS's single-threaded model. + compareAndDelete: (ns, key) => + Effect.sync(() => { + const id = k(ns, key); + if (!store.has(id)) return false; + store.delete(id); + return true; + }), }; }; @@ -243,6 +273,37 @@ export const makeFumaBlobStore = (fuma: IFumaClient): BlobStore => ({ (cause) => new StorageError({ message: "FumaDB blob operation failed", cause }), ), ), + compareAndDelete: (namespace, key) => + fuma + .transaction( + Effect.gen(function* () { + const id = blobId(namespace, key); + // Read inside the transaction: on libSQL/Postgres, `fuma.transaction` + // runs real BEGIN/COMMIT, so concurrent transactions serialize and + // no other fiber can interleave between this get and the delete — + // exactly one caller observes a present row, everyone else sees + // absent-after-commit. FumaDB's query builder discards rows-affected + // counts (deleteMany -> Promise) and exposes no raw driver + // handle, so a single `DELETE ... RETURNING` statement is not + // reachable through this abstraction without a cross-host driver + // change; the serializing transaction is the equivalent guarantee + // here. (The in-memory store's synchronous Map op is the atomic + // counterpart.) + const row = (yield* fuma.use("blob.cad.find", (db) => + db.findFirst("blob", { where: (b) => b("id", "=", id) }), + )) as BlobRow | null; + if (row === null) return false; + yield* fuma.use("blob.cad.delete", (db) => + db.deleteMany("blob", { where: (b) => b("id", "=", id) }), + ); + return true; + }), + ) + .pipe( + Effect.mapError( + (cause) => new StorageError({ message: "FumaDB blob operation failed", cause }), + ), + ), has: (namespace, key) => fuma .use("blob.has", (db) => diff --git a/packages/core/sdk/src/pending-approval.ts b/packages/core/sdk/src/pending-approval.ts index 72c76fa7ef..226218a4c4 100644 --- a/packages/core/sdk/src/pending-approval.ts +++ b/packages/core/sdk/src/pending-approval.ts @@ -18,8 +18,9 @@ // to resuming the fiber, without requiring the fiber to still exist. // // Records live in the existing owner-scoped `blob` table (no new table, no -// migration). They are single-use: consumed on resume, so one approval authorizes -// exactly one invocation and a replayed resume cannot re-run the call. +// migration). They are single-use: consumed atomically (compareAndDelete) on +// resume, so one approval authorizes exactly one invocation and NEITHER a +// replayed resume NOR two concurrent resumes can re-run the call. // --------------------------------------------------------------------------- import { Effect, Option, Schema } from "effect"; @@ -93,11 +94,24 @@ export const makePendingApprovalStore = ( consume: (executionId) => Effect.gen(function* () { + // Read the payload first — compareAndDelete returns only a boolean, + // so the winner needs the record's value to validate and return. + // Reading before the gate is safe: a concurrent loser may read the + // same payload but will fail the compareAndDelete gate below and + // never return it. The ordering that matters is the DELETE's + // single-winner guarantee, not the read's. const raw = yield* blobs.get(namespace, executionId); if (raw === null) return null; - // Consume before validating: a record we are about to reject is a record - // nobody should be able to retry against. - yield* blobs.delete(namespace, executionId); + // Atomic single-winner delete: exactly one concurrent consumer + // observes true (the record existed and was removed); everyone else + // observes false. This closes the get→delete race that previously + // let two concurrent resumes both read a record before either + // deleted it — duplicated side effects. + const removed = yield* blobs.compareAndDelete(namespace, executionId); + if (!removed) return null; + // Consume before validating (preserved): a record we are about to + // reject is a record nobody should be able to retry against — the + // atomic delete already removed it, so no other consumer can see it. const decoded = decodePendingApproval(raw); if (Option.isNone(decoded)) return null; const approval = decoded.value; diff --git a/packages/hosts/cloudflare/src/blob-store.ts b/packages/hosts/cloudflare/src/blob-store.ts index 631718a3ce..1ccfc8292a 100644 --- a/packages/hosts/cloudflare/src/blob-store.ts +++ b/packages/hosts/cloudflare/src/blob-store.ts @@ -5,8 +5,7 @@ // lives here so the SDK stays platform-agnostic. // // Object name: `${namespace}/${key}`. Unambiguous because a namespace is -// always `partition/pluginId` (exactly one slash; partitions use `:` -// separators, plugin ids contain no slash), so the first two segments always +// always `partition/pluginId` (exactly one slash; partitions use `:// separators, plugin ids contain no slash), so the first two segments always // recover the namespace and the rest is the key. // // Unlike `makeFumaBlobStore`, writes do NOT participate in FumaDB @@ -25,46 +24,82 @@ const objectName = (namespace: string, key: string): string => `${namespace}/${k const storeError = (op: string) => (cause: unknown) => new StorageError({ message: `R2 blob ${op} failed`, cause }); -export const makeR2BlobStore = (bucket: R2Bucket): BlobStore => ({ - get: (namespace, key) => - Effect.tryPromise({ - try: async () => { - const object = await bucket.get(objectName(namespace, key)); - return object == null ? null : await object.text(); - }, - catch: storeError("get"), - }), - // R2 has no multi-get; fetch the (at most two — user + org partition) - // namespaces concurrently. - getMany: (namespaces, key) => - Effect.tryPromise({ - try: async () => { - const hits = new Map(); - await Promise.all( - namespaces.map(async (namespace) => { - const object = await bucket.get(objectName(namespace, key)); - if (object != null) hits.set(namespace, await object.text()); +export const makeR2BlobStore = (bucket: R2Bucket): BlobStore => { + // Claims for compareAndDelete: at most one in-flight claim per object in + // this isolate. JavaScript is single-threaded per isolate, so the + // synchronous has+add below is atomic and the only interleaving risk is + // between the await points of the head+delete pair — the claim gate + // closes exactly that window (a second fiber sees the claim and observes + // false without touching R2). Cross-isolate races remain + // last-writer-wins: R2 offers no conditional delete. The caller pattern + // (single consume per approval; idempotent re-consume returns null) + // tolerates that residual, same as the orphaned-write caveat above. + const claims = new Set(); + + return { + get: (namespace, key) => + Effect.tryPromise({ + try: async () => { + const object = await bucket.get(objectName(namespace, key)); + return object == null ? null : await object.text(); + }, + catch: storeError("get"), + }), + // R2 has no multi-get; fetch the (at most two — user + org partition) + // namespaces concurrently. + getMany: (namespaces, key) => + Effect.tryPromise({ + try: async () => { + const hits = new Map(); + await Promise.all( + namespaces.map(async (namespace) => { + const object = await bucket.get(objectName(namespace, key)); + if (object != null) hits.set(namespace, await object.text()); + }), + ); + return hits; + }, + catch: storeError("getMany"), + }), + put: (namespace, key, value) => + Effect.tryPromise({ + try: async () => { + await bucket.put(objectName(namespace, key), value); + }, + catch: storeError("put"), + }), + delete: (namespace, key) => + Effect.tryPromise({ + try: () => bucket.delete(objectName(namespace, key)), + catch: storeError("delete"), + }), + has: (namespace, key) => + Effect.tryPromise({ + try: async () => (await bucket.head(objectName(namespace, key))) != null, + catch: storeError("has"), + }), + compareAndDelete: (namespace, key) => { + const name = objectName(namespace, key); + // Gate 1: isolate-level claim, atomic (synchronous has+add). + if (claims.has(name)) return Effect.succeed(false); + claims.add(name); + return Effect.tryPromise({ + try: async () => { + // Gate 2: existence check at delete time. An absent object means + // someone else removed it — this caller loses. + const head = await bucket.head(name); + if (head == null) return false; + await bucket.delete(name); + return true; + }, + catch: storeError("compareAndDelete"), + }).pipe( + Effect.ensuring( + Effect.sync(() => { + claims.delete(name); }), - ); - return hits; - }, - catch: storeError("getMany"), - }), - put: (namespace, key, value) => - Effect.tryPromise({ - try: async () => { - await bucket.put(objectName(namespace, key), value); - }, - catch: storeError("put"), - }), - delete: (namespace, key) => - Effect.tryPromise({ - try: () => bucket.delete(objectName(namespace, key)), - catch: storeError("delete"), - }), - has: (namespace, key) => - Effect.tryPromise({ - try: async () => (await bucket.head(objectName(namespace, key))) != null, - catch: storeError("has"), - }), -}); + ), + ); + }, + }; +};