Skip to content

Commit c3fc5cd

Browse files
committed
fix(sdk): make pending-approval consumption atomic
1 parent 521c77b commit c3fc5cd

5 files changed

Lines changed: 307 additions & 49 deletions

File tree

‎.changeset/approval-atomicity.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
---
2+
"@executor-js/sdk": patch
3+
---
4+
5+
fix: make pending-approval consumption atomic
6+
7+
`PendingApprovalStore.consume` previously read the record and deleted it as
8+
two separate operations. Two concurrent resumes (a double-click, a client
9+
retry, or two hosts racing the same approval) could both read the record
10+
before either deleted it, and both would execute the approved tool call —
11+
duplicated side effects from a single approval.
12+
13+
Consumption now goes through a new `BlobStore.compareAndDelete` primitive
14+
with a single-winner guarantee: exactly one concurrent consumer observes the
15+
record as present-and-removed; everyone else observes it as absent. The
16+
in-memory store implements it as a synchronous Map operation (atomic in JS's
17+
single-threaded model); the FumaDB-backed store implements it as
18+
get+delete inside the serializing transaction the driver already provides
19+
(libSQL/Postgres BEGIN/COMMIT). The approval's expiry and corrupt-record
20+
semantics are unchanged.
Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
import { describe, expect, it } from "@effect/vitest";
2+
import { Effect, Predicate } from "effect";
3+
4+
import { makeInMemoryBlobStore, type BlobStore } from "./blob";
5+
import {
6+
makePendingApprovalStore,
7+
PENDING_APPROVAL_TTL_MS,
8+
type PendingApproval,
9+
} from "./pending-approval";
10+
11+
// ---------------------------------------------------------------------------
12+
// Focused tests — pending-approval consume atomicity, deterministic layer.
13+
//
14+
// These tests pin the exactly-once consume invariant deterministically,
15+
// plus the expiry/corrupt semantics required to survive the consume
16+
// restructure. A companion fast-check property (concurrent consumes ⇒
17+
// exactly one non-null) lives elsewhere.
18+
//
19+
// NOTE on test infrastructure:
20+
// - it.effect runs under @effect/vitest's TestContext scheduler, where
21+
// Effect.sleep never advances — any async boundary deadlocks the test.
22+
// Sync-only effects work; async ones must go through Effect.runPromise in
23+
// a plain vitest test.
24+
// - The concurrency proof needs the real scheduler AND an explicit
25+
// read-barrier: without one, the synchronous Map store completes each
26+
// consume before the next fiber starts, so the double-resume race can
27+
// never manifest and the test would be vacuous as a concurrency proof.
28+
// ---------------------------------------------------------------------------
29+
30+
const approval = (overrides?: Partial<PendingApproval>): PendingApproval => ({
31+
executionId: "exec_1",
32+
artifactId: "art_1",
33+
code: 'return await tools.github.user.main.issues.create({"title":"x"})',
34+
address: "github.user.main.issues.create",
35+
expiresAt: Date.now() + PENDING_APPROVAL_TTL_MS,
36+
...overrides,
37+
});
38+
39+
describe("approval consume atomicity (compareAndDelete gate)", () => {
40+
it("exactly one of N concurrent consumes wins (single-winner invariant)", async () => {
41+
const N = 8;
42+
const backing = makeInMemoryBlobStore();
43+
// Promise-latch barrier: every fiber increments the arrival count after
44+
// its read and parks on the shared promise, which resolves only when
45+
// all N have arrived — every read happens before any delete. Pure JS
46+
// promise semantics, immune to Effect scheduler/runtime differences.
47+
let arrived = 0;
48+
let releaseAll!: () => void;
49+
const allArrived = new Promise<void>((resolve) => {
50+
releaseAll = resolve;
51+
});
52+
const barrier: BlobStore = {
53+
...backing,
54+
get: (ns, key) =>
55+
backing.get(ns, key).pipe(
56+
Effect.tap(() =>
57+
Effect.sync(() => {
58+
arrived++;
59+
if (arrived === N) releaseAll();
60+
}),
61+
),
62+
// tap (not andThen — that would discard the payload): park until
63+
// every fiber has read, then pass the payload through untouched.
64+
Effect.tap(() => Effect.promise(() => allArrived)),
65+
),
66+
};
67+
68+
const outcome = await Effect.runPromise(
69+
Effect.gen(function* () {
70+
const store = makePendingApprovalStore(barrier, "u:t:s");
71+
yield* store.put(approval());
72+
73+
const results = yield* Effect.all(
74+
Array.from({ length: N }, () => store.consume("exec_1")),
75+
{ concurrency: "unbounded" },
76+
);
77+
const winners = results.filter(Predicate.isNotNull);
78+
79+
// Post-condition: the record is gone for everyone.
80+
const after = yield* store.consume("exec_1");
81+
return { winners: winners.map((w) => w.address), after };
82+
}),
83+
);
84+
85+
expect(outcome.winners.length).toBe(1);
86+
expect(outcome.winners[0]).toBe("github.user.main.issues.create");
87+
expect(outcome.after).toBeNull();
88+
});
89+
90+
it.effect("a replayed consume after a win reads absent", () =>
91+
Effect.gen(function* () {
92+
const store = makePendingApprovalStore(makeInMemoryBlobStore(), "u:t:s");
93+
yield* store.put(approval());
94+
expect(yield* store.consume("exec_1")).not.toBeNull();
95+
expect(yield* store.consume("exec_1")).toBeNull();
96+
expect(yield* store.consume("exec_1")).toBeNull();
97+
}),
98+
);
99+
100+
it.effect("expired records are consumed-and-dropped (never retried)", () =>
101+
Effect.gen(function* () {
102+
const blobs = makeInMemoryBlobStore();
103+
let now = 1_000_000;
104+
const store = makePendingApprovalStore(blobs, "u:t:s", () => now);
105+
yield* store.put(approval({ expiresAt: now + 10 }));
106+
107+
now += 100; // expires
108+
expect(yield* store.consume("exec_1")).toBeNull();
109+
110+
// The record is gone — rolling the clock back does not resurrect it.
111+
now = 1_000_000;
112+
expect(yield* store.consume("exec_1")).toBeNull();
113+
}),
114+
);
115+
116+
it.effect("corrupt records are consumed-and-dropped (never surfaced)", () =>
117+
Effect.gen(function* () {
118+
const blobs = makeInMemoryBlobStore();
119+
yield* blobs.put("u:t:s/@pending-approval", "exec_1", "not-json{{");
120+
121+
const store = makePendingApprovalStore(blobs, "u:t:s");
122+
expect(yield* store.consume("exec_1")).toBeNull();
123+
124+
// Gone for good — a second consume finds nothing.
125+
expect(yield* store.consume("exec_1")).toBeNull();
126+
}),
127+
);
128+
});

‎packages/core/sdk/src/blob.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,26 @@ export interface BlobStore {
3838
) => Effect.Effect<void, StorageError>;
3939
readonly delete: (namespace: string, key: string) => Effect.Effect<void, StorageError>;
4040
readonly has: (namespace: string, key: string) => Effect.Effect<boolean, StorageError>;
41+
/**
42+
* Atomically delete a record IF it exists. Returns true iff this caller's
43+
* delete removed an existing record; false iff it was already absent.
44+
*
45+
* The single-winner invariant: exactly one concurrent caller observes true
46+
* for a given (namespace, key); everyone else observes false, and the
47+
* post-condition is that the record is absent for all callers. There is no
48+
* read-your-undefined window — a caller that observed true is guaranteed
49+
* the record was present at deletion time, and no other caller can observe
50+
* true for the same key afterwards.
51+
*
52+
* Implementations MUST NOT do check-then-act across two statements:
53+
* either a single atomic statement with a rows-affected/count check, or
54+
* get+delete inside a serializing transaction (libSQL/Postgres
55+
* `fuma.transaction`), or a single synchronous Map op (in-memory).
56+
*/
57+
readonly compareAndDelete: (
58+
namespace: string,
59+
key: string,
60+
) => Effect.Effect<boolean, StorageError>;
4161
}
4262

4363
export interface PluginBlobStore {
@@ -154,6 +174,16 @@ export const makeInMemoryBlobStore = (): BlobStore => {
154174
store.delete(k(ns, key));
155175
}),
156176
has: (ns, key) => Effect.sync(() => store.has(k(ns, key))),
177+
// Atomic by construction: a synchronous Map has+delete runs to completion
178+
// without yielding, so no other fiber can interleave between the check
179+
// and the delete in JS's single-threaded model.
180+
compareAndDelete: (ns, key) =>
181+
Effect.sync(() => {
182+
const id = k(ns, key);
183+
if (!store.has(id)) return false;
184+
store.delete(id);
185+
return true;
186+
}),
157187
};
158188
};
159189

@@ -243,6 +273,37 @@ export const makeFumaBlobStore = (fuma: IFumaClient): BlobStore => ({
243273
(cause) => new StorageError({ message: "FumaDB blob operation failed", cause }),
244274
),
245275
),
276+
compareAndDelete: (namespace, key) =>
277+
fuma
278+
.transaction(
279+
Effect.gen(function* () {
280+
const id = blobId(namespace, key);
281+
// Read inside the transaction: on libSQL/Postgres, `fuma.transaction`
282+
// runs real BEGIN/COMMIT, so concurrent transactions serialize and
283+
// no other fiber can interleave between this get and the delete —
284+
// exactly one caller observes a present row, everyone else sees
285+
// absent-after-commit. FumaDB's query builder discards rows-affected
286+
// counts (deleteMany -> Promise<void>) and exposes no raw driver
287+
// handle, so a single `DELETE ... RETURNING` statement is not
288+
// reachable through this abstraction without a cross-host driver
289+
// change; the serializing transaction is the equivalent guarantee
290+
// here. (The in-memory store's synchronous Map op is the atomic
291+
// counterpart.)
292+
const row = (yield* fuma.use("blob.cad.find", (db) =>
293+
db.findFirst("blob", { where: (b) => b("id", "=", id) }),
294+
)) as BlobRow | null;
295+
if (row === null) return false;
296+
yield* fuma.use("blob.cad.delete", (db) =>
297+
db.deleteMany("blob", { where: (b) => b("id", "=", id) }),
298+
);
299+
return true;
300+
}),
301+
)
302+
.pipe(
303+
Effect.mapError(
304+
(cause) => new StorageError({ message: "FumaDB blob operation failed", cause }),
305+
),
306+
),
246307
has: (namespace, key) =>
247308
fuma
248309
.use("blob.has", (db) =>

‎packages/core/sdk/src/pending-approval.ts‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,9 @@
1818
// to resuming the fiber, without requiring the fiber to still exist.
1919
//
2020
// Records live in the existing owner-scoped `blob` table (no new table, no
21-
// migration). They are single-use: consumed on resume, so one approval authorizes
22-
// exactly one invocation and a replayed resume cannot re-run the call.
21+
// migration). They are single-use: consumed atomically (compareAndDelete) on
22+
// resume, so one approval authorizes exactly one invocation and NEITHER a
23+
// replayed resume NOR two concurrent resumes can re-run the call.
2324
// ---------------------------------------------------------------------------
2425

2526
import { Effect, Option, Schema } from "effect";
@@ -93,11 +94,24 @@ export const makePendingApprovalStore = (
9394

9495
consume: (executionId) =>
9596
Effect.gen(function* () {
97+
// Read the payload first — compareAndDelete returns only a boolean,
98+
// so the winner needs the record's value to validate and return.
99+
// Reading before the gate is safe: a concurrent loser may read the
100+
// same payload but will fail the compareAndDelete gate below and
101+
// never return it. The ordering that matters is the DELETE's
102+
// single-winner guarantee, not the read's.
96103
const raw = yield* blobs.get(namespace, executionId);
97104
if (raw === null) return null;
98-
// Consume before validating: a record we are about to reject is a record
99-
// nobody should be able to retry against.
100-
yield* blobs.delete(namespace, executionId);
105+
// Atomic single-winner delete: exactly one concurrent consumer
106+
// observes true (the record existed and was removed); everyone else
107+
// observes false. This closes the get→delete race that previously
108+
// let two concurrent resumes both read a record before either
109+
// deleted it — duplicated side effects.
110+
const removed = yield* blobs.compareAndDelete(namespace, executionId);
111+
if (!removed) return null;
112+
// Consume before validating (preserved): a record we are about to
113+
// reject is a record nobody should be able to retry against — the
114+
// atomic delete already removed it, so no other consumer can see it.
101115
const decoded = decodePendingApproval(raw);
102116
if (Option.isNone(decoded)) return null;
103117
const approval = decoded.value;

0 commit comments

Comments
 (0)