diff --git a/packages/subscriptions/src/channels/slack-store.ts b/packages/subscriptions/src/channels/slack-store.ts index b7e9b1e40..d3973d663 100644 --- a/packages/subscriptions/src/channels/slack-store.ts +++ b/packages/subscriptions/src/channels/slack-store.ts @@ -29,6 +29,15 @@ 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. + releaseAnchorClaim( + event: SlackThreadEvent, + subscriberId: number, + ): Promise; // Atomically claim delivery of (event, subscriber, update). 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 @@ -47,6 +56,21 @@ export interface SlackAnchorStore { } 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. +const CLAIM_TTL_SECONDS = 60; + +const ANCHOR_CLAIM = { pending: true } as const; +type SlackAnchorClaim = typeof ANCHOR_CLAIM; + +// Delete only while the key still holds the claim, never a real anchor. +// 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]) +end +return 0 +`; function anchorKey(event: SlackThreadEvent, subscriberId: number): string { return `slack:${event.kind}:${event.id}:sub:${subscriberId}`; @@ -72,10 +96,11 @@ function getRedis(): Redis { export function createRedisAnchorStore(): SlackAnchorStore { return { async getAnchor(event, subscriberId) { - const raw = await getRedis().get( + const raw = await getRedis().get( anchorKey(event, subscriberId), ); - return raw ?? null; + if (!raw || "pending" in raw) return null; + return raw; }, async setAnchor(event, subscriberId, anchor) { await getRedis().set(anchorKey(event, subscriberId), anchor, { @@ -85,6 +110,21 @@ export function createRedisAnchorStore(): SlackAnchorStore { async clearAnchor(event, subscriberId) { 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"; + }, + async releaseAnchorClaim(event, subscriberId) { + await getRedis().eval( + RELEASE_CLAIM, + [anchorKey(event, subscriberId)], + [JSON.stringify(ANCHOR_CLAIM)], + ); + }, async reserveDelivery(event, subscriberId, updateId) { const res = await getRedis().set( deliveredKey(event, subscriberId, updateId), @@ -101,16 +141,30 @@ export function createRedisAnchorStore(): SlackAnchorStore { export function createMemoryAnchorStore(): SlackAnchorStore { const anchors = new Map(); + const claims = new Set(); const delivered = new Set(); return { async getAnchor(event, subscriberId) { return anchors.get(anchorKey(event, subscriberId)) ?? null; }, async setAnchor(event, subscriberId, anchor) { - anchors.set(anchorKey(event, subscriberId), anchor); + const key = anchorKey(event, subscriberId); + claims.delete(key); + anchors.set(key, anchor); }, async clearAnchor(event, subscriberId) { - anchors.delete(anchorKey(event, subscriberId)); + const key = anchorKey(event, subscriberId); + claims.delete(key); + anchors.delete(key); + }, + async claimAnchor(event, subscriberId) { + const key = anchorKey(event, subscriberId); + if (anchors.has(key) || claims.has(key)) return false; + claims.add(key); + return true; + }, + async releaseAnchorClaim(event, subscriberId) { + claims.delete(anchorKey(event, subscriberId)); }, 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 495a7f1f8..b5c02180b 100644 --- a/packages/subscriptions/src/channels/slack.test.ts +++ b/packages/subscriptions/src/channels/slack.test.ts @@ -328,6 +328,131 @@ describe("createSlackChannel", () => { expect((await store.getAnchor(REPORT, 1))?.ts).toBe("1700000000.0001"); expect((await store.getAnchor(MAINTENANCE, 1))?.ts).toBe("1700000000.0002"); }); + + for (const { kind, status } of [ + { kind: REPORT, status: "investigating" }, + { kind: MAINTENANCE, status: "maintenance" }, + ] as const) { + test(`concurrent updates of one ${kind.kind} open a single root`, async () => { + // The winner's root post blocks until the loser starts waiting, so both + // deliveries are in flight before any anchor exists. + let openGate = () => {}; + const gate = new Promise((resolve) => { + openGate = resolve; + }); + const { client, calls } = makeClient(); + let roots = 0; + const gatedClient: SlackClient = { + ...client, + async postMessage(args) { + if (!args.thread_ts) { + // A second root means the race is back; don't hang the test on it. + roots += 1; + if (roots > 1) openGate(); + await gate; + } + return client.postMessage(args); + }, + }; + let sleeps = 0; + const store = createMemoryAnchorStore(); + const channel = createSlackChannel({ + store, + createClient: () => gatedClient, + getBotToken: token, + softUnsubscribe: noopUnsub, + sleep: async () => { + sleeps += 1; + openGate(); + await new Promise((resolve) => setTimeout(resolve, 0)); + }, + }); + + await Promise.all([ + channel.sendNotifications( + [makeSub()], + makeUpdate({ status, updateId: 100 }), + ), + channel.sendNotifications( + [makeSub()], + makeUpdate({ status, updateId: 101 }), + ), + ]); + + const posts = calls.filter((c) => c.method === "post"); + expect(sleeps).toBeGreaterThan(0); + expect(posts.filter((c) => c.thread_ts === undefined).length).toBe(1); + // Loser backfills the winner's update, then replies with its own. + expect( + posts.filter((c) => c.thread_ts === "1700000000.0001").length, + ).toBe(2); + expect(calls.filter((c) => c.method === "update").length).toBe(1); + expect((await store.getAnchor(kind, 1))?.ts).toBe("1700000000.0001"); + }); + } + + test("failed root post releases the claim so a retry opens the thread", async () => { + let attempts = 0; + const roots: (string | undefined)[] = []; + const client: SlackClient = { + async postMessage(args) { + attempts += 1; + roots.push(args.thread_ts); + if (attempts === 1) { + const error = new Error("slack error") as Error & { + data?: { error?: string }; + }; + error.data = { error: "ratelimited" }; + throw error; + } + return { ts: "1700000000.0001" }; + }, + async update(args) { + return { ts: args.ts }; + }, + }; + let sleeps = 0; + const store = createMemoryAnchorStore(); + const channel = createSlackChannel({ + store, + createClient: () => client, + getBotToken: token, + softUnsubscribe: noopUnsub, + sleep: async () => { + sleeps += 1; + }, + }); + + await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); + expect(await store.getAnchor(REPORT, 1)).toBeNull(); + await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); + + expect(sleeps).toBe(0); + expect(roots).toEqual([undefined, undefined]); + expect((await store.getAnchor(REPORT, 1))?.ts).toBe("1700000000.0001"); + }); + + test("claim held for the whole wait releases the delivery without posting", async () => { + const { client, calls } = makeClient(); + let sleeps = 0; + const store = createMemoryAnchorStore(); + const channel = createSlackChannel({ + store, + createClient: () => client, + getBotToken: token, + softUnsubscribe: noopUnsub, + sleep: async () => { + sleeps += 1; + }, + }); + expect(await store.claimAnchor(REPORT, 1)).toBe(true); + + await channel.sendNotifications([makeSub()], makeUpdate({ updateId: 100 })); + + expect(calls.length).toBe(0); + expect(sleeps).toBeGreaterThan(0); + expect(await store.reserveDelivery(REPORT, 1, 100)).toBe(true); + }); }); describe("validateSlackConfig", () => { @@ -375,3 +500,30 @@ describe("reserveDelivery (memory store)", () => { expect(await store.reserveDelivery(REPORT, 2, 3)).toBe(true); }); }); + +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.getAnchor(REPORT, 2)).toBeNull(); + // Kinds are namespaced: a maintenance with the same id claims separately. + expect(await store.claimAnchor(MAINTENANCE, 2)).toBe(true); + }); + + test("setAnchor replaces the claim; a set anchor can't be claimed", async () => { + const store = createMemoryAnchorStore(); + expect(await store.claimAnchor(REPORT, 2)).toBe(true); + await store.setAnchor(REPORT, 2, { ts: "1.0", channelId: "C1" }); + await store.releaseAnchorClaim(REPORT, 2); + expect((await store.getAnchor(REPORT, 2))?.ts).toBe("1.0"); + expect(await store.claimAnchor(REPORT, 2)).toBe(false); + }); + + test("releaseAnchorClaim makes the anchor claimable again", 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); + }); +}); diff --git a/packages/subscriptions/src/channels/slack.ts b/packages/subscriptions/src/channels/slack.ts index ef6208c0e..eb8bcc54a 100644 --- a/packages/subscriptions/src/channels/slack.ts +++ b/packages/subscriptions/src/channels/slack.ts @@ -10,6 +10,7 @@ import type { PageUpdate, Subscription } from "../types"; import { buildReplyMessage, buildRootMessage } from "./slack-blocks"; import { type SlackAnchorStore, + type SlackThreadAnchor, type SlackThreadEvent, createRedisAnchorStore, } from "./slack-store"; @@ -28,8 +29,16 @@ export interface SlackChannelDeps { createClient: (token: string) => SlackClient; getBotToken: (teamId: string) => Promise; softUnsubscribe: (subscriberId: number) => Promise; + sleep?: (ms: number) => Promise; } +// A concurrent delivery that lost the root claim waits for the winner's root +// (one postMessage round trip). Slack deliveries are not retried by the +// 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. const CHANNEL_TERMINAL_ERRORS = new Set([ @@ -99,6 +108,25 @@ async function softUnsubscribeInDb(subscriberId: number): Promise { } export function createSlackChannel(deps: SlackChannelDeps) { + const sleep = + 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. + async function acquireAnchor( + event: SlackThreadEvent, + subscriberId: number, + ): 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; + } + return null; + } + async function runSlack( subscriberId: number, fn: () => Promise, @@ -150,10 +178,17 @@ export function createSlackChannel(deps: SlackChannelDeps) { // failed post we release it below so the delivery stays retriable. if (!(await deps.store.reserveDelivery(event, sub.id, updateId))) return; - const anchor = await deps.store.getAnchor(event, sub.id); + const anchor = await acquireAnchor(event, sub.id); + if (!anchor) { + console.error( + `slack: ${event.kind} ${event.id} root still in flight for subscriber ${sub.id} — dropping update ${updateId}`, + ); + await deps.store.releaseDelivery(event, sub.id, updateId); + return; + } const root = buildRootMessage(pageUpdate, sub); - if (!anchor) { + if (anchor === CLAIMED) { const res = await runSlack(sub.id, () => client.postMessage({ channel: channelId, @@ -161,18 +196,21 @@ export function createSlackChannel(deps: SlackChannelDeps) { }), ); if (!res || res === TEAM_TOKEN_INVALID) { + await deps.store.releaseAnchorClaim(event, sub.id); await deps.store.releaseDelivery(event, sub.id, updateId); return res === TEAM_TOKEN_INVALID ? TEAM_TOKEN_INVALID : undefined; } - if (res.ts) { - // Stash this first update so the next one can backfill it into the - // thread before the root is re-rendered and its content lost. - await deps.store.setAnchor(event, sub.id, { - ts: res.ts, - channelId, - pendingRootReply: buildReplyMessage(pageUpdate), - }); + if (!res.ts) { + await deps.store.releaseAnchorClaim(event, sub.id); + return; } + // Stash this first update so the next one can backfill it into the + // thread before the root is re-rendered and its content lost. + await deps.store.setAnchor(event, sub.id, { + ts: res.ts, + channelId, + pendingRootReply: buildReplyMessage(pageUpdate), + }); return; }