diff --git a/services/notifications/src/__tests__/dispatch-push.test.ts b/services/notifications/src/__tests__/dispatch-push.test.ts index 7c8ea0432b..50c78491a9 100644 --- a/services/notifications/src/__tests__/dispatch-push.test.ts +++ b/services/notifications/src/__tests__/dispatch-push.test.ts @@ -1039,27 +1039,34 @@ describe('NotificationChannelDO.dispatchPush — local push sink', () => { const payload = sinkCall?.[1] as { idempotencyKey: string; payload: { - title: string; - body: string; - data: unknown; + titleLength: number; + bodyLength: number; + dataKeys: string[]; + dataType: unknown; sound: string | null; priority: string; }; to: string; }; expect(payload.idempotencyKey).toBe(idemKey); + // Content-free: sizes + key names + discriminator, never the values. expect(payload.payload).toEqual({ - title: 'My title', - body: 'My body', - data: { type: 'cloud_agent_session', cliSessionId: 'ses_sink' }, + titleLength: 'My title'.length, + bodyLength: 'My body'.length, + dataKeys: ['cliSessionId', 'type'], + dataType: 'cloud_agent_session', sound: 'default', priority: 'high', }); expect(payload.to).toBe(''); - // Token must never appear in the sink payload. + // Neither the device token nor the user/agent-controlled content may appear. const serialized = JSON.stringify(payload); expect(serialized).not.toContain('tok-real'); expect(serialized).not.toContain('ExponentPushToken'); + // User/agent-controlled content values must never appear (identifiers such + // as the idempotency key are fine and intentionally retained for tracing). + expect(serialized).not.toContain('My title'); + expect(serialized).not.toContain('My body'); // No receipt work created. expect(env.RECEIPTS_QUEUE.send).not.toHaveBeenCalled(); @@ -1159,4 +1166,55 @@ describe('NotificationChannelDO.dispatchPush — local push sink', () => { expect(logSpy.mock.calls.some(call => call[0] === 'agent_push_sink_payload')).toBe(false); logSpy.mockRestore(); }); + + it('does not double-send when the same key is dispatched concurrently (in-flight claim)', async () => { + installDbMock({ tokens: [{ user_id: 'user-concurrent', token: 'tok1' }] }); + vi.spyOn(env.EVENT_SERVICE, 'isUserInContext').mockResolvedValue(false); + + // Hold the first Expo send open so a concurrent replay of the same key + // interleaves while the first attempt is mid-flight — i.e. after it wrote + // its `pending` marker but before it recorded a terminal state. Without the + // in-flight claim the replay would read `pending`, treat it as a reusable + // retry slot, and submit a second OS push. + let releaseExpo: () => void = () => undefined; + const expoGate = new Promise(resolve => { + releaseExpo = resolve; + }); + vi.mocked(sendPushNotifications).mockImplementationOnce(async () => { + await expoGate; + return { + ticketTokenPairs: [{ ticketId: 't1', token: 'tok1' }], + staleTokens: [], + ticketErrors: [], + }; + }); + + const stub = getDO('user-concurrent'); + const input: DispatchPushInput = { + userId: 'user-concurrent', + presenceContext: null, + idempotencyKey: 'agent-notification:ses_cc:n-cc', + badge: null, + push: { + title: 'T', + body: 'B', + data: { type: 'cloud_agent_session', cliSessionId: 'ses_cc' }, + sound: 'default', + priority: 'high', + }, + }; + + const [first, second] = await runInDurableObject(stub, async instance => { + const p1 = instance.dispatchPush(input); + // Let p1 progress to the awaiting-Expo point before the replay enters. + await new Promise(resolve => setTimeout(resolve, 0)); + const p2 = instance.dispatchPush(input); + releaseExpo(); + return Promise.all([p1, p2]); + }); + + expect([first.kind, second.kind].sort()).toEqual(['delivered', 'duplicate']); + // The core guarantee: exactly one OS push submission for the key. + expect(sendPushNotifications).toHaveBeenCalledTimes(1); + }); }); diff --git a/services/notifications/src/dos/NotificationChannelDO.ts b/services/notifications/src/dos/NotificationChannelDO.ts index 5813c57144..6bb4d8b545 100644 --- a/services/notifications/src/dos/NotificationChannelDO.ts +++ b/services/notifications/src/dos/NotificationChannelDO.ts @@ -43,7 +43,29 @@ const IDEM_TTL_MS = 60 * 60 * 1000; // 1 hour const ACCEPTED_BOOKKEEPING_RETRY_DELAY_MS = 30_000; export class NotificationChannelDO extends DurableObject { + // In-flight claims by idempotency key. The DO serializes turns, but a turn + // yields the isolate on every external `await` (presence RPC, Expo send), so a + // concurrent replay of the same key can interleave: it would read the storage + // `pending` marker the first attempt left before its send, treat it as a + // reusable retry slot, and send a second time. This in-memory claim closes + // that window — a replay seeing a live claim returns `duplicate`, while a + // sequential post-failure retry (claim already released) still reuses the + // `pending` slot as before. + private readonly inFlight = new Set(); + async dispatchPush(input: DispatchPushInput): Promise { + if (this.inFlight.has(input.idempotencyKey)) { + return { kind: 'duplicate' }; + } + this.inFlight.add(input.idempotencyKey); + try { + return await this.dispatchPushInner(input); + } finally { + this.inFlight.delete(input.idempotencyKey); + } + } + + private async dispatchPushInner(input: DispatchPushInput): Promise { // 1. Idempotency. DO is single-threaded — requests for a given // user serialize on this instance. Retryable send failures leave the // record at `pending` so upstream can retry the send without @@ -160,12 +182,22 @@ export class NotificationChannelDO extends DurableObject { // means the sink captured the payload, never device delivery. if (isPushSinkEnabled(this.env)) { const ts = Date.now(); + // Content-free observability only. Title, body, and the `data` object are + // agent/user-controlled and can carry credentials or private customer + // content; logging their values would retain that in local, CI, and + // aggregated Worker logs. Record sizes, `data` key names, and the fixed + // `type` discriminator instead — enough to trace the send without content. + const data = input.push.data; console.log('agent_push_sink_payload', { idempotencyKey: input.idempotencyKey, payload: { - title: input.push.title, - body: input.push.body, - data: input.push.data, + titleLength: input.push.title.length, + bodyLength: input.push.body.length, + dataKeys: data && typeof data === 'object' ? Object.keys(data).sort() : [], + dataType: + data && typeof data === 'object' && 'type' in data + ? (data as { type?: unknown }).type + : undefined, sound: input.push.sound ?? null, priority: input.push.priority ?? 'default', },