From c3e6fe7457fe53374f0f55365d15a2f1f056cf26 Mon Sep 17 00:00:00 2001 From: Maximilian Kaske Date: Thu, 15 Oct 2026 11:31:37 +0200 Subject: [PATCH] fix(subscriptions): token-scoped anchor claims and single root backfill - claimAnchor returns a per-claim token; releaseAnchorClaim only deletes the claim it owns, so an expired claimant can't free a newer one - reserve the root backfill per root ts, so losers woken together don't all post the first update into the thread - document why the claim TTL stays short --- .../subscriptions/src/channels/slack-store.ts | 71 +++++++++++-------- .../subscriptions/src/channels/slack.test.ts | 64 ++++++++++++++--- packages/subscriptions/src/channels/slack.ts | 24 ++++--- 3 files changed, 110 insertions(+), 49 deletions(-) diff --git a/packages/subscriptions/src/channels/slack-store.ts b/packages/subscriptions/src/channels/slack-store.ts index d3973d663..bc8ccc7d1 100644 --- a/packages/subscriptions/src/channels/slack-store.ts +++ b/packages/subscriptions/src/channels/slack-store.ts @@ -29,42 +29,52 @@ export interface SlackAnchorStore { anchor: SlackThreadAnchor, ): Promise; clearAnchor(event: SlackThreadEvent, subscriberId: number): Promise; - // Atomically claim the right to post the thread root. Returns true only for - // the caller that wins; the winner must `setAnchor` or `releaseAnchorClaim`. - // `getAnchor` reads a held claim as null. - claimAnchor(event: SlackThreadEvent, subscriberId: number): Promise; - // Drop a claim whose root post failed. No-op once a real anchor is set. + // Atomically claim the right to post the thread root. Returns a token only + // for the caller that wins; the winner must `setAnchor` or + // `releaseAnchorClaim`. `getAnchor` reads a held claim as null. + claimAnchor( + event: SlackThreadEvent, + subscriberId: number, + ): Promise; + // Drop a claim whose root post failed. No-op unless the key still holds + // this token, so an expired claimant can't release a newer claim. releaseAnchorClaim( event: SlackThreadEvent, subscriberId: number, + token: string, ): Promise; - // Atomically claim delivery of (event, subscriber, update). Returns true only + // Atomically claim delivery of (event, subscriber, update). `update` is an + // update id, or a string key for a delivery that isn't one (root backfill). Returns true only // for the caller that wins the claim; concurrent callers get false and must // skip. This is the dedupe reservation — a single atomic op, not a // read-then-write pair, so two dispatchers can't both post the same message. reserveDelivery( event: SlackThreadEvent, subscriberId: number, - updateId: number, + updateId: number | string, ): Promise; // Release a reservation whose post failed, so the delivery can be retried. releaseDelivery( event: SlackThreadEvent, subscriberId: number, - updateId: number, + updateId: number | string, ): Promise; } const TTL_SECONDS = 90 * 24 * 60 * 60; -// Short so a winner that crashes mid-post doesn't block the thread for long; -// `setAnchor` replaces the claim with the full TTL. +// Short so a winner that dies mid-post doesn't block the thread for long; +// `setAnchor` replaces the claim with the full TTL. A root post retried by the +// Slack client past this can open a second root — preferred over losers +// dropping their updates for the length of the client's retry window. const CLAIM_TTL_SECONDS = 60; -const ANCHOR_CLAIM = { pending: true } as const; -type SlackAnchorClaim = typeof ANCHOR_CLAIM; +interface SlackAnchorClaim { + pending: string; +} -// Delete only while the key still holds the claim, never a real anchor. -// Upstash JSON-serializes values, so ARGV[1] is the stringified claim. +// Delete only while the key still holds this claim, never a real anchor or a +// newer claim. Upstash JSON-serializes values, so ARGV[1] is the stringified +// claim. const RELEASE_CLAIM = ` if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) @@ -79,7 +89,7 @@ function anchorKey(event: SlackThreadEvent, subscriberId: number): string { function deliveredKey( event: SlackThreadEvent, subscriberId: number, - updateId: number, + updateId: number | string, ): string { return `${anchorKey(event, subscriberId)}:update:${updateId}`; } @@ -111,18 +121,19 @@ export function createRedisAnchorStore(): SlackAnchorStore { await getRedis().del(anchorKey(event, subscriberId)); }, async claimAnchor(event, subscriberId) { - const res = await getRedis().set( - anchorKey(event, subscriberId), - ANCHOR_CLAIM, - { ex: CLAIM_TTL_SECONDS, nx: true }, - ); - return res === "OK"; + const claim: SlackAnchorClaim = { pending: crypto.randomUUID() }; + const res = await getRedis().set(anchorKey(event, subscriberId), claim, { + ex: CLAIM_TTL_SECONDS, + nx: true, + }); + return res === "OK" ? claim.pending : null; }, - async releaseAnchorClaim(event, subscriberId) { + async releaseAnchorClaim(event, subscriberId, token) { + const claim: SlackAnchorClaim = { pending: token }; await getRedis().eval( RELEASE_CLAIM, [anchorKey(event, subscriberId)], - [JSON.stringify(ANCHOR_CLAIM)], + [JSON.stringify(claim)], ); }, async reserveDelivery(event, subscriberId, updateId) { @@ -141,7 +152,7 @@ export function createRedisAnchorStore(): SlackAnchorStore { export function createMemoryAnchorStore(): SlackAnchorStore { const anchors = new Map(); - const claims = new Set(); + const claims = new Map(); const delivered = new Set(); return { async getAnchor(event, subscriberId) { @@ -159,12 +170,14 @@ export function createMemoryAnchorStore(): SlackAnchorStore { }, async claimAnchor(event, subscriberId) { const key = anchorKey(event, subscriberId); - if (anchors.has(key) || claims.has(key)) return false; - claims.add(key); - return true; + if (anchors.has(key) || claims.has(key)) return null; + const token = crypto.randomUUID(); + claims.set(key, token); + return token; }, - async releaseAnchorClaim(event, subscriberId) { - claims.delete(anchorKey(event, subscriberId)); + async releaseAnchorClaim(event, subscriberId, token) { + const key = anchorKey(event, subscriberId); + if (claims.get(key) === token) claims.delete(key); }, async reserveDelivery(event, subscriberId, updateId) { const key = deliveredKey(event, subscriberId, updateId); diff --git a/packages/subscriptions/src/channels/slack.test.ts b/packages/subscriptions/src/channels/slack.test.ts index b5c02180b..de78424d4 100644 --- a/packages/subscriptions/src/channels/slack.test.ts +++ b/packages/subscriptions/src/channels/slack.test.ts @@ -391,6 +391,46 @@ describe("createSlackChannel", () => { }); } + test("losers woken together backfill the first update once", async () => { + let openGate = () => {}; + const gate = new Promise((resolve) => { + openGate = resolve; + }); + const { client, calls } = makeClient(); + const gatedClient: SlackClient = { + ...client, + async postMessage(args) { + // Thread posts yield too, so woken losers interleave mid-backfill. + if (args.thread_ts) + await new Promise((resolve) => setTimeout(resolve, 0)); + else await gate; + return client.postMessage(args); + }, + }; + const store = createMemoryAnchorStore(); + const channel = createSlackChannel({ + store, + createClient: () => gatedClient, + getBotToken: token, + softUnsubscribe: noopUnsub, + sleep: async () => { + openGate(); + await new Promise((resolve) => setTimeout(resolve, 0)); + }, + }); + + await Promise.all( + [100, 101, 102].map((updateId) => + channel.sendNotifications([makeSub()], makeUpdate({ updateId })), + ), + ); + + const posts = calls.filter((c) => c.method === "post"); + expect(posts.filter((c) => c.thread_ts === undefined).length).toBe(1); + // One backfill of the root's update plus one reply per loser. + expect(posts.filter((c) => c.thread_ts !== undefined).length).toBe(3); + }); + test("failed root post releases the claim so a retry opens the thread", async () => { let attempts = 0; const roots: (string | undefined)[] = []; @@ -445,7 +485,7 @@ describe("createSlackChannel", () => { sleeps += 1; }, }); - expect(await store.claimAnchor(REPORT, 1)).toBe(true); + expect(await store.claimAnchor(REPORT, 1)).not.toBeNull(); await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); @@ -504,26 +544,28 @@ describe("reserveDelivery (memory store)", () => { describe("claimAnchor (memory store)", () => { test("only the first claim wins and reads as no anchor", async () => { const store = createMemoryAnchorStore(); - expect(await store.claimAnchor(REPORT, 2)).toBe(true); - expect(await store.claimAnchor(REPORT, 2)).toBe(false); + expect(await store.claimAnchor(REPORT, 2)).not.toBeNull(); + expect(await store.claimAnchor(REPORT, 2)).toBeNull(); expect(await store.getAnchor(REPORT, 2)).toBeNull(); // Kinds are namespaced: a maintenance with the same id claims separately. - expect(await store.claimAnchor(MAINTENANCE, 2)).toBe(true); + expect(await store.claimAnchor(MAINTENANCE, 2)).not.toBeNull(); }); test("setAnchor replaces the claim; a set anchor can't be claimed", async () => { const store = createMemoryAnchorStore(); - expect(await store.claimAnchor(REPORT, 2)).toBe(true); + const token = await store.claimAnchor(REPORT, 2); await store.setAnchor(REPORT, 2, { ts: "1.0", channelId: "C1" }); - await store.releaseAnchorClaim(REPORT, 2); + await store.releaseAnchorClaim(REPORT, 2, token ?? ""); expect((await store.getAnchor(REPORT, 2))?.ts).toBe("1.0"); - expect(await store.claimAnchor(REPORT, 2)).toBe(false); + expect(await store.claimAnchor(REPORT, 2)).toBeNull(); }); - test("releaseAnchorClaim makes the anchor claimable again", async () => { + test("releaseAnchorClaim frees only its own claim", async () => { const store = createMemoryAnchorStore(); - expect(await store.claimAnchor(REPORT, 2)).toBe(true); - await store.releaseAnchorClaim(REPORT, 2); - expect(await store.claimAnchor(REPORT, 2)).toBe(true); + const token = await store.claimAnchor(REPORT, 2); + await store.releaseAnchorClaim(REPORT, 2, "stale-token"); + expect(await store.claimAnchor(REPORT, 2)).toBeNull(); + await store.releaseAnchorClaim(REPORT, 2, token ?? ""); + expect(await store.claimAnchor(REPORT, 2)).not.toBeNull(); }); }); diff --git a/packages/subscriptions/src/channels/slack.ts b/packages/subscriptions/src/channels/slack.ts index eb8bcc54a..e8ba8d672 100644 --- a/packages/subscriptions/src/channels/slack.ts +++ b/packages/subscriptions/src/channels/slack.ts @@ -37,7 +37,6 @@ export interface SlackChannelDeps { // dispatcher, so giving up here drops the update. const ANCHOR_WAIT_ATTEMPTS = 10; const ANCHOR_WAIT_MS = 500; -const CLAIMED = Symbol("claimed"); // Channel-scoped terminal errors: this subscriber's destination is gone for // good, so we stop delivering to that one subscriber. @@ -112,17 +111,18 @@ export function createSlackChannel(deps: SlackChannelDeps) { deps.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); - // Existing anchor, CLAIMED when this caller must post the root, or null if - // another caller held the claim for the whole wait. + // Existing anchor, a claim token when this caller must post the root, or + // null if another caller held the claim for the whole wait. async function acquireAnchor( event: SlackThreadEvent, subscriberId: number, - ): Promise { + ): Promise { for (let attempt = 0; attempt < ANCHOR_WAIT_ATTEMPTS; attempt++) { if (attempt > 0) await sleep(ANCHOR_WAIT_MS); const anchor = await deps.store.getAnchor(event, subscriberId); if (anchor) return anchor; - if (await deps.store.claimAnchor(event, subscriberId)) return CLAIMED; + const claim = await deps.store.claimAnchor(event, subscriberId); + if (claim) return { claim }; } return null; } @@ -188,7 +188,7 @@ export function createSlackChannel(deps: SlackChannelDeps) { } const root = buildRootMessage(pageUpdate, sub); - if (anchor === CLAIMED) { + if ("claim" in anchor) { const res = await runSlack(sub.id, () => client.postMessage({ channel: channelId, @@ -196,12 +196,12 @@ export function createSlackChannel(deps: SlackChannelDeps) { }), ); if (!res || res === TEAM_TOKEN_INVALID) { - await deps.store.releaseAnchorClaim(event, sub.id); + await deps.store.releaseAnchorClaim(event, sub.id, anchor.claim); await deps.store.releaseDelivery(event, sub.id, updateId); return res === TEAM_TOKEN_INVALID ? TEAM_TOKEN_INVALID : undefined; } if (!res.ts) { - await deps.store.releaseAnchorClaim(event, sub.id); + await deps.store.releaseAnchorClaim(event, sub.id, anchor.claim); return; } // Stash this first update so the next one can backfill it into the @@ -216,7 +216,12 @@ export function createSlackChannel(deps: SlackChannelDeps) { // The first update only ever lived in the root. Before we overwrite the // root below, post it into the thread so its history is preserved. - if (anchor.pendingRootReply) { + // Reserved per root: losers woken together all see `pendingRootReply`. + const backfillKey = `root:${anchor.ts}`; + if ( + anchor.pendingRootReply && + (await deps.store.reserveDelivery(event, sub.id, backfillKey)) + ) { const backfill = anchor.pendingRootReply; const backfillRes = await runSlack(sub.id, () => client.postMessage({ @@ -227,6 +232,7 @@ export function createSlackChannel(deps: SlackChannelDeps) { }), ); if (!backfillRes || backfillRes === TEAM_TOKEN_INVALID) { + await deps.store.releaseDelivery(event, sub.id, backfillKey); await deps.store.releaseDelivery(event, sub.id, updateId); return backfillRes === TEAM_TOKEN_INVALID ? TEAM_TOKEN_INVALID -- 2.51.2