fix(web): make the draft claim exclusive with a Durable Object

Claims on a community-agent draft (post or discard) now go through the
DraftClaimLock Durable Object when the DRAFT_CLAIM_LOCK binding is present:
one SQLite-backed object per draft identity (idFromName), whose one-event-
at-a-time input gate makes grant/refuse/release atomic. The lease expires
after 15 minutes so a crashed or unknown-outcome post cannot wedge the
draft, and after a recorded decision the object keeps refusing claims for
2 minutes while the KV decision marker propagates to other locations. The
KV get/put claim stays as the fallback for next dev and tests.

A second discard of the same draft (double click) now returns 200 whether
the first one has finished or is still holding the draft.

worker.ts exports the class (OpenNext custom-worker pattern); wrangler.jsonc
adds the binding and a v1 new_sqlite_classes migration, applied on deploy.

Tests: draft-claim-lock (6) + 7 new route tests with a serializing fake DO
(concurrent posts -> one 200, stale-KV retry refused during the hold, lease
kept after 502 and expired after 15 min, 422 frees at once, double discard
-> 200/200, post/discard exclusion, unreachable lock fails closed). Against
7cf21caee, 6 of the 7 route tests fail (the 422 release test passes on
both, a guard against over-locking).
vitest community-agent*/content-watch/public-api-security/draft-claim-lock/
deploy-preflight: 7 files, 89 passed. tsc --noEmit exit 0. eslint on touched
files exit 0. esbuild bundles worker.ts with the exported class.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_014ZwqatxgVFxHvovngywnks
This commit is contained in:
Hunter BandClaude Opus 5.5 committed 2026-09-29 12:25:41 -07:00
1 parent 7cf21caeed
commit ccc0009881
8 files changed
+531 -60

No files matched your search

+9 -5
View File
@@ -165,21 +165,24 @@ export async function POST(req: Request) {
if (draft.posted || resolution?.state === "posted" || resolution?.state === "posting") {
return NextResponse.json({ error: "draft already posted" }, { status: 409 });
}
// A repeated discard (double click) is a no-op, not an error.
// A repeated discard (double click) is a no-op, not an error: whether the
// first one already finished or is still running.
if (resolution?.state === "discarded") return discarded();
// Hold the same claim a post takes, so a discard and a post of one draft
// do not both run.
let result: DraftClaimResult;
try {
result = await claimDraft(env.CURATED_KV, parsedKey.type, parsedKey.id);
result = await claimDraft(env, parsedKey.type, parsedKey.id, "discard");
} catch (e) {
return NextResponse.json({ error: `could not claim draft: ${String(e)}` }, { status: 500 });
}
if (!result.ok) {
if (result.reason === "resolved" && result.resolution.state === "discarded") return discarded();
if (result.reason === "held" && result.holder === "discard") return discarded();
return claimRefused(result);
}
const claim = result.claim;
let recorded = false;
try {
// The marker stops the next cron run from regenerating this draft.
await markDraftResolved(env.CURATED_KV, parsedKey.type, parsedKey.id, "discarded");
@@ -187,11 +190,12 @@ export async function POST(req: Request) {
if (draft.type === "digest") {
await deleteDigestRecord(env.CURATED_KV, draft.id);
}
recorded = true;
} catch (e) {
return NextResponse.json({ error: `discard failed: ${String(e)}` }, { status: 500 });
} finally {
// The marker (or, if it failed, the draft) now carries the decision.
await releaseDraftClaim(env.CURATED_KV, claim).catch(() => undefined);
await releaseDraftClaim(env.CURATED_KV, claim, { recorded }).catch(() => undefined);
}
if (draft.type === "digest") revalidateDigestPage();
return discarded();
@@ -217,7 +221,7 @@ export async function POST(req: Request) {
}
let result: DraftClaimResult;
try {
result = await claimDraft(env.CURATED_KV, parsedKey.type, parsedKey.id);
result = await claimDraft(env, parsedKey.type, parsedKey.id, "post");
} catch (e) {
return NextResponse.json({ error: `could not claim draft: ${String(e)}` }, { status: 500 });
}
@@ -240,7 +244,7 @@ export async function POST(req: Request) {
await markDraftResolved(env.CURATED_KV, parsedKey.type, parsedKey.id, "posted");
// The marker now carries the decision, so a later claim sees it; a
// reopened draft (new activity clears the marker) is postable again.
await releaseDraftClaim(env.CURATED_KV, heldClaim).catch(() => undefined);
await releaseDraftClaim(env.CURATED_KV, heldClaim, { recorded: true }).catch(() => undefined);
await env.CURATED_KV?.put(draftKey, JSON.stringify(draft), { expirationTtl: 60 * 60 * 24 * 7 });
return undefined;
} catch (e) {
+149 -27
View File
@@ -32,6 +32,7 @@ import {
type AgentDraft,
} from "./community-agent";
import { runDigest, runDupes, runTriage } from "./community-agent-tasks";
import { FakeDraftClaimLock } from "./draft-claim-lock.fake";
/** In-memory KV that pages like Cloudflare KV (max 1000 keys per list call). */
class FakeKv {
@@ -138,9 +139,10 @@ async function act(kv: FakeKv, fields: Record<string, unknown>): Promise<Respons
return adminPost(adminRequest(postBody({ ...fields, reviewedSha256 })));
}
function useAdminEnv(kv: FakeKv) {
function stubAdminEnv(kv: FakeKv, lock?: FakeDraftClaimLock) {
mocks.getAgentEnv.mockResolvedValue({
CURATED_KV: kv,
...(lock ? { DRAFT_CLAIM_LOCK: lock } : {}),
MAINTAINER_TOKEN: "configured",
MAINTAINER_GITHUB_PAT: "ghp_test",
GITHUB_REPO: "Hmbown/CodeWhale",
@@ -213,7 +215,7 @@ describe("weekly digest publication requires maintainer approval", () => {
expect(staged).toMatchObject({ approved: false });
expect(isPublishedDigest(staged)).toBe(false);
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
const draftKey = onlyKey(kv, "draft:digest:");
const res = await act(kv, { action: "post", draftKey, lang: "en" });
@@ -231,7 +233,7 @@ describe("weekly digest publication requires maintainer approval", () => {
mocks.agentChat.mockResolvedValue({ content: JSON.stringify(DIGEST_MODEL_OUTPUT), usage: { input: 1, output: 1 } });
await runDigest({ CURATED_KV: kv, DEEPSEEK_API_KEY: "k" });
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const draftKey = onlyKey(kv, "draft:digest:");
const res = await act(kv, { action: "post", draftKey, lang: "zh" });
@@ -245,7 +247,7 @@ describe("weekly digest publication requires maintainer approval", () => {
mocks.agentChat.mockResolvedValue({ content: JSON.stringify(DIGEST_MODEL_OUTPUT), usage: { input: 1, output: 1 } });
await runDigest({ CURATED_KV: kv, DEEPSEEK_API_KEY: "k" });
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const draftKey = onlyKey(kv, "draft:digest:");
expect((await act(kv, { action: "post", draftKey, lang: "en" })).status).toBe(200);
@@ -261,7 +263,7 @@ describe("weekly digest publication requires maintainer approval", () => {
mocks.agentChat.mockResolvedValue({ content: JSON.stringify(DIGEST_MODEL_OUTPUT), usage: { input: 1, output: 1 } });
await runDigest({ CURATED_KV: kv, DEEPSEEK_API_KEY: "k" });
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const draftKey = onlyKey(kv, "draft:digest:");
const res = await act(kv, { action: "post", draftKey, lang: "en", editedBody: "# Edited" });
@@ -280,7 +282,7 @@ describe("weekly digest publication requires maintainer approval", () => {
mocks.agentChat.mockResolvedValue({ content: JSON.stringify(DIGEST_MODEL_OUTPUT), usage: { input: 1, output: 1 } });
await runDigest({ CURATED_KV: kv, DEEPSEEK_API_KEY: "k" });
useAdminEnv(kv);
stubAdminEnv(kv);
const draftKey = onlyKey(kv, "draft:digest:");
const res = await act(kv, { action: "discard", draftKey });
await expect(res.json()).resolves.toMatchObject({ ok: true, action: "discarded" });
@@ -298,7 +300,7 @@ describe("weekly digest publication requires maintainer approval", () => {
mocks.agentChat.mockResolvedValue({ content: JSON.stringify(DIGEST_MODEL_OUTPUT), usage: { input: 1, output: 1 } });
await runDigest({ CURATED_KV: kv, DEEPSEEK_API_KEY: "k" });
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub(422);
const draftKey = onlyKey(kv, "draft:digest:");
const res = await act(kv, { action: "post", draftKey, lang: "en" });
@@ -329,7 +331,7 @@ describe("weekly digest publication requires maintainer approval", () => {
expect(JSON.parse(kv.values.get(draftKey)!).bodyEn).toContain("Digest B");
expect(kv.values.get(recordKey)).toBe(recordA);
useAdminEnv(kv);
stubAdminEnv(kv);
const bodies: string[] = [];
vi.stubGlobal("fetch", vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
bodies.push(String(init?.body));
@@ -355,7 +357,7 @@ describe("weekly digest publication requires maintainer approval", () => {
const recordKey = onlyKey(kv, "digest:weekly-");
kv.values.set(recordKey, JSON.stringify({ ...JSON.parse(kv.values.get(recordKey)!), extra: "<script>" }));
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const draftKey = onlyKey(kv, "draft:digest:");
await expect((await act(kv, { action: "post", draftKey, lang: "zh" })).json()).resolves.toMatchObject({
@@ -404,7 +406,7 @@ describe("resolved drafts are not resurrected by the cron", () => {
it("does not redraft a discarded triage item", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const res = await act(kv, { action: "discard", draftKey: "draft:triage:42" });
expect(res.status).toBe(200);
@@ -417,7 +419,7 @@ describe("resolved drafts are not resurrected by the cron", () => {
it("does not turn a posted triage draft back into a pending one after the issue updates", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const res = await act(kv, { action: "post", draftKey: "draft:triage:42", lang: "en" });
expect(res.status).toBe(200);
@@ -432,7 +434,7 @@ describe("resolved drafts are not resurrected by the cron", () => {
it("drafts again when the item has new activity well after the maintainer's decision", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
expect((await act(kv, { action: "post", draftKey: "draft:triage:42" })).status).toBe(200);
@@ -467,7 +469,7 @@ describe("admin post action is idempotent and bounded", () => {
it("refuses to post a draft that is already posted", async () => {
const kv = new FakeKv();
kv.values.set("draft:triage:42", JSON.stringify(draft({ posted: true })));
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
const res = await act(kv, { action: "post", draftKey: "draft:triage:42" });
@@ -478,7 +480,7 @@ describe("admin post action is idempotent and bounded", () => {
it("returns ok when saving state fails after GitHub accepted the comment, and a retry does not post twice", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
kv.failPutsMatching = /^draft:triage:42$/;
@@ -494,7 +496,7 @@ describe("admin post action is idempotent and bounded", () => {
it("releases the claim when GitHub rejects the post so the maintainer can retry", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub(422);
const failed = await act(kv, { action: "post", draftKey: "draft:triage:42" });
@@ -511,7 +513,7 @@ describe("admin post action is idempotent and bounded", () => {
async (status) => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub(status);
const failed = await act(kv, { action: "post", draftKey: "draft:triage:42" });
@@ -529,7 +531,7 @@ describe("admin post action is idempotent and bounded", () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
kv.yieldEachOp = true;
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
const results = await Promise.all([
@@ -591,7 +593,7 @@ describe("admin post action is idempotent and bounded", () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
kv.yieldEachOp = true;
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = raceAfterAbsenceChecks(kv);
const results = await Promise.all([
@@ -607,7 +609,7 @@ describe("admin post action is idempotent and bounded", () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
kv.yieldEachOp = true;
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = raceAfterAbsenceChecks(kv);
const [post, discard] = await Promise.all([
@@ -625,7 +627,7 @@ describe("admin post action is idempotent and bounded", () => {
it("claims without KV list, whose results can lag writes by up to a minute", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
// A stale listing that never shows claim keys must not affect the claim.
const realList = kv.list.bind(kv);
@@ -643,7 +645,7 @@ describe("admin post action is idempotent and bounded", () => {
it("answers a claim it cannot read back as retryable, not as a lost race, and the retry posts", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
// The first read-back of the request's own claim misses the write.
const realGet = kv.get.bind(kv);
@@ -666,7 +668,7 @@ describe("admin post action is idempotent and bounded", () => {
it("refuses a post while another request's claim is visible, and leaves that claim alone", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
kv.values.set("draft-claim:triage:42", "other-request");
@@ -681,7 +683,7 @@ describe("admin post action is idempotent and bounded", () => {
it("treats a repeated discard as a no-op, including one that sees the marker only at claim time", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
stubGitHub();
const first = await act(kv, { action: "discard", draftKey: "draft:triage:42" });
@@ -711,7 +713,7 @@ describe("admin post action is idempotent and bounded", () => {
it("refuses an action on a draft regenerated after the page loaded, before calling GitHub", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft({ bodyEn: "reviewed A" }));
useAdminEnv(kv);
stubAdminEnv(kv);
const shownA = await reviewedBodyHash("reviewed A");
// The cron regenerates the pending draft while the admin page shows A.
await saveDraft(kv, draft({ bodyEn: "unseen B", generatedAt: "2026-01-02T00:00:00.000Z" }));
@@ -730,7 +732,7 @@ describe("admin post action is idempotent and bounded", () => {
it("requires the reviewed-text hash", async () => {
const kv = new FakeKv();
await saveDraft(kv, draft());
useAdminEnv(kv);
stubAdminEnv(kv);
const posts = stubGitHub();
const res = await adminPost(adminRequest(postBody({ action: "post", draftKey: "draft:triage:42" })));
@@ -740,7 +742,7 @@ describe("admin post action is idempotent and bounded", () => {
it("answers malformed JSON with a JSON 400", async () => {
const kv = new FakeKv();
useAdminEnv(kv);
stubAdminEnv(kv);
const res = await adminPost(adminRequest("{not json"));
expect(res.status).toBe(400);
@@ -749,7 +751,7 @@ describe("admin post action is idempotent and bounded", () => {
it("counts streamed body bytes instead of trusting a missing Content-Length", async () => {
const kv = new FakeKv();
useAdminEnv(kv);
stubAdminEnv(kv);
const chunk = new TextEncoder().encode("x".repeat(40_000));
const stream = new ReadableStream<Uint8Array>({
start(controller) {
@@ -764,6 +766,126 @@ describe("admin post action is idempotent and bounded", () => {
});
});
describe("admin claims with the DRAFT_CLAIM_LOCK Durable Object bound", () => {
const KEY = "draft:triage:42";
async function lockedAdmin(status = 201) {
const kv = new FakeKv();
await saveDraft(kv, draft());
kv.yieldEachOp = true;
const lock = new FakeDraftClaimLock();
stubAdminEnv(kv, lock);
const claimKeysWritten: string[] = [];
kv.beforePut = async (key) => {
if (key.startsWith("draft-claim:")) claimKeysWritten.push(key);
};
const posts = stubGitHub(status);
return { kv, lock, posts, claimKeysWritten };
}
/** What a location whose KV has not yet seen the first request's writes reads. */
function forgetDecision(kv: FakeKv) {
kv.values.set(KEY, JSON.stringify(draft()));
kv.values.delete("draft-resolved:triage:42");
}
it("lets exactly one of several concurrent posts through, without KV claim keys", async () => {
const { kv, lock, posts, claimKeysWritten } = await lockedAdmin();
const results = await Promise.all(
Array.from({ length: 4 }, () => act(kv, { action: "post", draftKey: KEY }))
);
expect(results.map((r) => r.status).sort()).toEqual([200, 409, 409, 409]);
expect(posts).toHaveLength(1);
expect(claimKeysWritten).toEqual([]);
expect(lock.calls.filter((c) => c.op === "claim")).toHaveLength(4);
});
it("refuses a retry that reads stale KV after a recorded post, until the hold ends", async () => {
const { kv, lock, posts } = await lockedAdmin();
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(200);
forgetDecision(kv);
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(409);
expect(posts).toHaveLength(1);
// Past the hold the KV marker has propagated; a location that really
// lacks it (a reopened draft) may act again.
lock.now += 2 * 60 * 1000;
forgetDecision(kv);
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(200);
expect(posts).toHaveLength(2);
});
it("keeps the lease after an unknown GitHub outcome, and lets it expire", async () => {
const { kv, lock, posts } = await lockedAdmin(502);
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(502);
stubGitHub();
// Even without the KV "posting" marker, the lease refuses a retry...
kv.values.delete("draft-resolved:triage:42");
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(409);
expect(posts).toHaveLength(1);
// ...until it expires, so the draft is not wedged for good.
lock.now += 15 * 60 * 1000;
const retryPosts = stubGitHub();
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(200);
expect(retryPosts).toHaveLength(1);
});
it("frees the draft at once when GitHub definitely rejected the post", async () => {
const { kv } = await lockedAdmin(422);
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(502);
const posts = stubGitHub();
expect((await act(kv, { action: "post", draftKey: KEY })).status).toBe(200);
expect(posts).toHaveLength(1);
});
it("treats a double-clicked discard as one discard, concurrent or repeated", async () => {
const { kv, posts } = await lockedAdmin();
const both = await Promise.all([
act(kv, { action: "discard", draftKey: KEY }),
act(kv, { action: "discard", draftKey: KEY }),
]);
expect(both.map((r) => r.status)).toEqual([200, 200]);
for (const res of both) await expect(res.json()).resolves.toEqual({ ok: true, action: "discarded" });
expect(JSON.parse(kv.values.get("draft-resolved:triage:42")!)).toMatchObject({ state: "discarded" });
// A third click served from stale KV hits the post-discard hold.
forgetDecision(kv);
expect((await act(kv, { action: "discard", draftKey: KEY })).status).toBe(200);
expect(posts).toEqual([]);
});
it("refuses a discard while a post holds the draft, and a post after a discard", async () => {
const { kv, posts } = await lockedAdmin();
const [post, discard] = await Promise.all([
act(kv, { action: "post", draftKey: KEY }),
act(kv, { action: "discard", draftKey: KEY }),
]);
expect([post.status, discard.status]).toEqual([200, 409]);
expect(posts).toHaveLength(1);
const other = await lockedAdmin();
expect((await act(other.kv, { action: "discard", draftKey: KEY })).status).toBe(200);
forgetDecision(other.kv);
expect((await act(other.kv, { action: "post", draftKey: KEY })).status).toBe(409);
expect(other.posts).toEqual([]);
});
it("fails closed when the lock cannot be reached", async () => {
const { kv, lock, posts } = await lockedAdmin();
lock.get = () => ({ act: async () => { throw new Error("DO unavailable"); } }) as never;
const res = await act(kv, { action: "post", draftKey: KEY });
expect(res.status).toBe(500);
expect(posts).toEqual([]);
expect(kv.values.get("draft-resolved:triage:42")).toBeUndefined();
});
});
describe("admin draft queue", () => {
function fill(kv: FakeKv, count: number) {
for (let i = 1; i <= count; i++) {
+97 -28
View File
@@ -9,6 +9,9 @@
* - Cites specific files / line numbers / linked issues when discussing code.
* - Always ends with the draft disclaimer.
*/
import type { DraftClaimLockNamespace, DraftClaimLockStub, DraftLockAction } from "./draft-claim-lock";
const MAX_OUTPUT_TOKENS = 2_000;
const FALLBACK_BASE = "https://api.deepseek.com";
const FALLBACK_MODEL = "deepseek-v4-flash";
@@ -244,6 +247,8 @@ interface KVNamespace {
export interface CommunityAgentEnv {
CURATED_KV?: KVNamespace;
/** Exclusive per-draft claim lock; see `claimDraft`. Absent locally. */
DRAFT_CLAIM_LOCK?: DraftClaimLockNamespace;
ADMIN_LOGIN_LIMITER?: { limit(options: { key: string }): Promise<{ success: boolean }> };
DEEPSEEK_API_KEY?: string;
DEEPSEEK_BASE_URL?: string;
@@ -259,7 +264,12 @@ export async function getAgentEnv(): Promise<CommunityAgentEnv> {
try {
const mod = await import("@opennextjs/cloudflare");
const ctx = await mod.getCloudflareContext({ async: true });
return ctx.env as CommunityAgentEnv;
const env = ctx.env as CommunityAgentEnv;
// `next dev` proxies bindings through wrangler, which cannot run a
// Durable Object class defined in this Worker's own entry; claims use
// the KV fallback there.
if (process.env.NODE_ENV === "development") return { ...env, DRAFT_CLAIM_LOCK: undefined };
return env;
} catch {
return {
DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY,
@@ -383,59 +393,105 @@ export async function markDraftResolved(
/**
* A hold on one draft identity while a maintainer action (post or discard)
* runs. It is a single KV key holding the claiming request's random token.
* runs. With the `DRAFT_CLAIM_LOCK` Durable Object bound it is a lease in
* that object (`lock` is set); without it, a KV key holding the claiming
* request's `<action>:<random>` token.
*/
export interface DraftClaim {
key: string;
token: string;
lock?: DraftClaimLockStub;
}
/**
* Why a claim was not taken:
* - `held`: another request's claim is visible (a post in flight, or one whose
* outcome is unknown, which holds the draft until the claim expires).
* - `held`: another request holds the draft (a post or discard in flight, one
* whose outcome is unknown, which holds it until the lease expires, or one
* that just finished). `holder` is that request's action, when known.
* - `resolved`: a decision (posted, discarded, a post in flight) exists.
* - `unconfirmed`: this request could not read its own claim back; nothing
* was done and a retry is safe.
* - `unconfirmed`: this request could not read its own KV claim back;
* nothing was done and a retry is safe. The Durable Object path never
* returns it.
*/
export type DraftClaimResult =
| { ok: true; claim: DraftClaim }
| { ok: false; reason: "held" | "unconfirmed" }
| { ok: false; reason: "held"; holder?: DraftLockAction }
| { ok: false; reason: "unconfirmed" }
| { ok: false; reason: "resolved"; resolution: DraftResolution };
/** Where claims live: the Durable Object when bound, else KV. */
export interface DraftClaimStores {
CURATED_KV?: KVNamespace;
DRAFT_CLAIM_LOCK?: DraftClaimLockNamespace;
}
const CLAIM_PREFIX = "draft-claim:";
// After an action whose decision is recorded, the Durable Object keeps
// refusing claims this long, so a retry in a location whose KV has not yet
// seen the decision marker (KV can lag 60 s or more) cannot act again.
const RECORDED_CLAIM_HOLD_MS = 2 * 60 * 1000;
function claimKey(type: AgentDraftType, id: string): string {
return CLAIM_PREFIX + draftKey(type, id).slice("draft:".length);
}
function claimHolder(token: string | null): DraftLockAction | undefined {
const action = token?.split(":", 1)[0];
return action === "post" || action === "discard" ? action : undefined;
}
/**
* Claim a draft before acting on it.
*
* This is best effort, not a lock. Workers KV has no compare-and-set: a
* write is usually visible at once to reads in the location that made it,
* but can take 60 seconds or more to reach other locations, and list results
* can lag even locally. So the claim uses only get and put on one key:
* refuse if a claim is already visible, write our token, and proceed only if
* reading the key back returns our token and no decision has been recorded.
* That stops the double-post cases this admin page actually produces (a
* second tab, a retry, a discard during a post), since each re-reads a key
* that a finished request wrote. Two requests whose writes land within the
* same moment, or that run in different locations, can still both proceed;
* a Durable Object would be needed to rule that out.
* With the `DRAFT_CLAIM_LOCK` Durable Object bound (production, once the
* binding and its migration are deployed) the claim is exclusive: one object
* per draft identity grants the lease to exactly one request, and the lease
* expires after 15 minutes so a crashed or unknown-outcome post cannot wedge
* the draft for good. After the lease, the KV decision marker is still
* checked, so a decision recorded after the caller's own check wins.
*
* A released claim can also stay visible to a retry from another location
* for up to about a minute, which surfaces as `held`.
* Without the binding (local dev, tests) the claim falls back to KV, which is
* best effort, not a lock. Workers KV has no compare-and-set: a write is
* usually visible at once to reads in the location that made it, but can
* take 60 seconds or more to reach other locations, and list results can lag
* even locally. So the fallback uses only get and put on one key: refuse if a
* claim is already visible, write our token, and proceed only if reading the
* key back returns our token and no decision has been recorded. That stops
* the sequential cases (a second tab, a retry, a discard during a post), but
* two requests whose writes land within the same moment, or that run in
* different locations, can both proceed.
*/
export async function claimDraft(
kv: KVNamespace | undefined,
stores: DraftClaimStores,
type: AgentDraftType,
id: string
id: string,
action: DraftLockAction
): Promise<DraftClaimResult> {
const token = crypto.randomUUID();
if (!kv) return { ok: true, claim: { key: "", token } };
const token = `${action}:${crypto.randomUUID()}`;
const key = claimKey(type, id);
if (await kv.get(key)) return { ok: false, reason: "held" };
const kv = stores.CURATED_KV;
if (stores.DRAFT_CLAIM_LOCK) {
const ns = stores.DRAFT_CLAIM_LOCK;
const lock = ns.get(ns.idFromName(key));
const granted = await lock.act({ op: "claim", token, action, leaseMs: POSTING_CLAIM_TTL_SEC * 1000 });
if (!granted.ok) return { ok: false, reason: "held", holder: granted.holder };
const claim: DraftClaim = { key, token, lock };
const drop = () => lock.act({ op: "release", token, holdMs: 0 }).catch(() => undefined);
const resolution = await getDraftResolution(kv, type, id).catch(async (e) => {
await drop();
throw e;
});
if (resolution) {
await drop();
return { ok: false, reason: "resolved", resolution };
}
return { ok: true, claim };
}
if (!kv) return { ok: true, claim: { key: "", token } };
const visible = await kv.get(key);
if (visible) return { ok: false, reason: "held", holder: claimHolder(visible) };
await kv.put(key, token, { expirationTtl: POSTING_CLAIM_TTL_SEC });
const seen = await kv.get(key);
if (seen !== token) {
@@ -445,7 +501,7 @@ export async function claimDraft(
await kv.delete(key).catch(() => undefined);
return { ok: false, reason: "unconfirmed" };
}
return { ok: false, reason: "held" };
return { ok: false, reason: "held", holder: claimHolder(seen) };
}
// A decision recorded after the caller's own check still wins.
const resolution = await getDraftResolution(kv, type, id).catch(async (e) => {
@@ -459,8 +515,21 @@ export async function claimDraft(
return { ok: true, claim: { key, token } };
}
/** Give up a claim, for an action that definitely did not happen or is recorded. */
export async function releaseDraftClaim(kv: KVNamespace | undefined, claim: DraftClaim): Promise<void> {
/**
* Give up a claim, for an action that definitely did not happen or whose
* decision is recorded. Pass `recorded: true` in the second case: the Durable
* Object then holds the draft a little longer (RECORDED_CLAIM_HOLD_MS) while
* the KV marker propagates, instead of freeing it at once.
*/
export async function releaseDraftClaim(
kv: KVNamespace | undefined,
claim: DraftClaim,
opts: { recorded?: boolean } = {}
): Promise<void> {
if (claim.lock) {
await claim.lock.act({ op: "release", token: claim.token, holdMs: opts.recorded ? RECORDED_CLAIM_HOLD_MS : 0 });
return;
}
if (!kv || !claim.key) return;
// Leave a claim another request has since written in place.
if ((await kv.get(claim.key)) !== claim.token) return;
+67
View File
@@ -0,0 +1,67 @@
/**
* Test double for the DRAFT_CLAIM_LOCK namespace. Each name gets its own
* storage, and each object runs one `act` at a time to completion, which is
* the guarantee a Durable Object's input gate gives the real class. Storage
* calls still yield to the event loop, so callers genuinely overlap.
*/
import {
applyDraftLock,
type DraftClaimLockId,
type DraftClaimLockNamespace,
type DraftClaimLockStub,
type DraftLockRequest,
type DraftLockResponse,
type DraftLockStorage,
} from "./draft-claim-lock";
class MapStorage implements DraftLockStorage {
readonly values = new Map<string, unknown>();
private async tick() {
await new Promise((resolve) => setTimeout(resolve, 0));
}
async get<T>(key: string) {
await this.tick();
return this.values.get(key) as T | undefined;
}
async put<T>(key: string, value: T) {
await this.tick();
this.values.set(key, structuredClone(value));
}
async delete(key: string) {
await this.tick();
return this.values.delete(key);
}
}
class FakeObject implements DraftClaimLockStub {
readonly storage = new MapStorage();
private queue: Promise<unknown> = Promise.resolve();
constructor(private readonly clock: () => number, private readonly calls: DraftLockRequest[]) {}
act(req: DraftLockRequest): Promise<DraftLockResponse> {
this.calls.push(req);
const run = this.queue.then(() => applyDraftLock(this.storage, this.clock(), req));
this.queue = run.catch(() => undefined);
return run;
}
}
export class FakeDraftClaimLock implements DraftClaimLockNamespace {
now = 1_700_000_000_000;
readonly calls: DraftLockRequest[] = [];
private readonly objects = new Map<string, FakeObject>();
idFromName(name: string): DraftClaimLockId {
return { toString: () => name };
}
get(id: DraftClaimLockId): FakeObject {
const name = id.toString();
let obj = this.objects.get(name);
if (!obj) {
obj = new FakeObject(() => this.now, this.calls);
this.objects.set(name, obj);
}
return obj;
}
}
+99
View File
@@ -0,0 +1,99 @@
import { describe, expect, it } from "vitest";
import { applyDraftLock, type DraftLockRequest, type DraftLockStorage } from "./draft-claim-lock";
import { FakeDraftClaimLock } from "./draft-claim-lock.fake";
/** Storage that yields on every call, so unserialized callers would interleave. */
class YieldingStorage implements DraftLockStorage {
readonly values = new Map<string, unknown>();
private async tick() {
await new Promise((resolve) => setTimeout(resolve, 0));
}
async get<T>(key: string) {
await this.tick();
return this.values.get(key) as T | undefined;
}
async put<T>(key: string, value: T) {
await this.tick();
this.values.set(key, value);
}
async delete(key: string) {
await this.tick();
return this.values.delete(key);
}
}
const claim = (token: string, action: "post" | "discard" = "post", leaseMs = 60_000): DraftLockRequest => ({
op: "claim",
token,
action,
leaseMs,
});
describe("DraftClaimLock", () => {
it("grants exactly one of many concurrent claims", async () => {
const ns = new FakeDraftClaimLock();
const lock = ns.get(ns.idFromName("draft-claim:triage:42"));
const results = await Promise.all(
Array.from({ length: 8 }, (_, i) => lock.act(claim(`post:${i}`)))
);
expect(results.filter((r) => r.ok)).toHaveLength(1);
expect(results.filter((r) => !r.ok)).toEqual(Array(7).fill({ ok: false, holder: "post" }));
});
it("without the object's one-event-at-a-time gate, the same logic would double-grant", async () => {
// Documents why the lock must run inside the Durable Object: the fake's
// serialization is what the input gate provides in production.
const storage = new YieldingStorage();
const results = await Promise.all([
applyDraftLock(storage, 0, claim("post:a")),
applyDraftLock(storage, 0, claim("post:b")),
]);
expect(results.every((r) => r.ok)).toBe(true);
});
it("keeps one lock per draft identity", async () => {
const ns = new FakeDraftClaimLock();
const a = ns.get(ns.idFromName("draft-claim:triage:1"));
const b = ns.get(ns.idFromName("draft-claim:triage:2"));
expect(await a.act(claim("post:a"))).toEqual({ ok: true });
expect(await b.act(claim("post:b"))).toEqual({ ok: true });
expect(await ns.get(ns.idFromName("draft-claim:triage:1")).act(claim("post:c"))).toEqual({
ok: false,
holder: "post",
});
});
it("lets a lease expire so a crashed action does not wedge the draft", async () => {
const ns = new FakeDraftClaimLock();
const lock = ns.get(ns.idFromName("k"));
expect(await lock.act(claim("post:a", "post", 15 * 60_000))).toEqual({ ok: true });
ns.now += 15 * 60_000 - 1;
expect(await lock.act(claim("discard:b", "discard"))).toEqual({ ok: false, holder: "post" });
ns.now += 1;
expect(await lock.act(claim("discard:b", "discard"))).toEqual({ ok: true });
// The expired holder can no longer release the new holder's lease.
await lock.act({ op: "release", token: "post:a", holdMs: 0 });
expect(await lock.act(claim("post:c"))).toEqual({ ok: false, holder: "discard" });
});
it("frees the draft on release, and only for the holder's own token", async () => {
const ns = new FakeDraftClaimLock();
const lock = ns.get(ns.idFromName("k"));
await lock.act(claim("post:a"));
await lock.act({ op: "release", token: "post:other", holdMs: 0 });
expect(await lock.act(claim("post:b"))).toEqual({ ok: false, holder: "post" });
await lock.act({ op: "release", token: "post:a", holdMs: 0 });
expect(await lock.act(claim("post:b"))).toEqual({ ok: true });
});
it("holds a recorded decision for holdMs, then frees it", async () => {
const ns = new FakeDraftClaimLock();
const lock = ns.get(ns.idFromName("k"));
await lock.act(claim("discard:a", "discard", 15 * 60_000));
await lock.act({ op: "release", token: "discard:a", holdMs: 120_000 });
ns.now += 119_999;
expect(await lock.act(claim("post:b"))).toEqual({ ok: false, holder: "discard" });
ns.now += 1;
expect(await lock.act(claim("post:b"))).toEqual({ ok: true });
});
});
+89
View File
@@ -0,0 +1,89 @@
/**
* The exclusive hold on one community-agent draft while a maintainer action
* (post or discard) runs.
*
* The `DraftClaimLock` Durable Object (exported from `worker.ts`, one instance
* per draft identity via `idFromName`) runs `applyDraftLock` against its own
* storage. A Durable Object handles one event at a time and its input gate
* holds other events while a storage call is pending, so the read and write
* below cannot interleave with another request's: exactly one claim wins.
*
* This module has no `cloudflare:workers` import so it runs under vitest and
* the Next.js build; `worker.ts` holds the thin class around it.
*/
export type DraftLockAction = "post" | "discard";
export type DraftLockRequest =
| { op: "claim"; token: string; action: DraftLockAction; leaseMs: number }
/**
* `holdMs` > 0 keeps refusing other claims for that long after an action
* whose decision was recorded, covering the time the KV decision marker
* takes to reach other locations. 0 frees the draft at once.
*/
| { op: "release"; token: string; holdMs: number };
export type DraftLockResponse =
| { ok: true }
| { ok: false; holder: DraftLockAction };
/** The subset of `DurableObjectStorage` the lock uses. */
export interface DraftLockStorage {
get<T>(key: string): Promise<T | undefined>;
put<T>(key: string, value: T): Promise<void>;
delete(key: string): Promise<boolean>;
}
interface Lease {
token: string;
action: DraftLockAction;
/** Epoch ms. A lease past this no longer holds, so a crashed action cannot wedge the draft. */
expiresAt: number;
}
const LEASE_KEY = "lease";
export async function applyDraftLock(
storage: DraftLockStorage,
now: number,
req: DraftLockRequest
): Promise<DraftLockResponse> {
const lease = await storage.get<Lease>(LEASE_KEY);
const live = lease && lease.expiresAt > now ? lease : undefined;
if (req.op === "claim") {
if (live && live.token !== req.token) return { ok: false, holder: live.action };
await storage.put<Lease>(LEASE_KEY, {
token: req.token,
action: req.action,
expiresAt: now + Math.max(0, req.leaseMs),
});
return { ok: true };
}
// Release: only the holder's own token can release, so a request whose
// lease expired and was retaken cannot free the new holder's claim.
if (!live || live.token !== req.token) return { ok: true };
if (req.holdMs > 0) {
await storage.put<Lease>(LEASE_KEY, { ...live, expiresAt: now + req.holdMs });
} else {
await storage.delete(LEASE_KEY);
}
return { ok: true };
}
/** The RPC surface of a `DraftClaimLock` stub. */
export interface DraftClaimLockStub {
act(req: DraftLockRequest): Promise<DraftLockResponse>;
}
/** The `DRAFT_CLAIM_LOCK` Durable Object namespace binding, as the app uses it. */
export interface DraftClaimLockNamespace {
idFromName(name: string): DraftClaimLockId;
get(id: DraftClaimLockId): DraftClaimLockStub;
}
/** Opaque `DurableObjectId`. */
export interface DraftClaimLockId {
toString(): string;
}
+13
View File
@@ -1,3 +1,4 @@
import { DurableObject } from "cloudflare:workers";
import handler from "./.open-next/worker.js";
import {
runCurate,
@@ -11,6 +12,18 @@ import {
import { runFactsDrift } from "./lib/facts-drift";
import { runLinkCheck, runSemanticDrift } from "./lib/content-watch";
import { fetchWithStaticInstaller } from "./lib/static-installer";
import { applyDraftLock, type DraftLockRequest } from "./lib/draft-claim-lock";
/**
* Exclusive claim on one community-agent draft (binding DRAFT_CLAIM_LOCK, one
* instance per draft via idFromName). OpenNext's custom-worker entry exports
* extra Durable Object classes from here; see lib/draft-claim-lock.ts.
*/
export class DraftClaimLock extends DurableObject {
async act(req: DraftLockRequest) {
return applyDraftLock(this.ctx.storage, Date.now(), req);
}
}
export default {
fetch(request, env, ctx) {
+8
View File
@@ -33,6 +33,14 @@
"id": "a2e6f324db9b4b03bbc940a4ba246985"
}
],
"durable_objects": {
"bindings": [
{ "name": "DRAFT_CLAIM_LOCK", "class_name": "DraftClaimLock" }
]
},
"migrations": [
{ "tag": "v1", "new_sqlite_classes": ["DraftClaimLock"] }
],
"vars": {
"GITHUB_REPO": "Hmbown/CodeWhale",
"DEEPSEEK_MODEL": "deepseek-v4-flash",