OUT-4105 | Stop dropping buffered grouped emails when the DB is briefly unreachable

The flush job's retry budget was ~3 seconds (3 attempts, 1s/2s backoff). When
the Supabase pooler went unreachable for 47 minutes on 8/24 every window
exhausted it on the first $queryRaw, and onFailure then DELETEd the whole
window — unsent rows included — so those grouped emails were destroyed, or
orphaned as sentAt IS NULL when the delete failed too. bufferGroupedEmailEvent
only reuses windows younger than 5 minutes, so orphans are never re-flushed.

Widen the in-run backoff to ~75s, and on exhaustion re-enqueue the window with
a 15m delay for up to two rounds before cleaning up. The run is already
idempotent (it reads sentAt IS NULL and marks each recipient as it goes), so a
re-enqueue only sends what is still outstanding. Sentry now fires once the
rounds are spent rather than on every window of a transient outage.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4tF84tzJW7Bo1DehsDX22

Author: priosshrsth

src/jobs/notifications/flush-grouped-email.test.ts

@@ -318,25 +318,41 @@ describe('flushGroupedEmailRun', () => {
 })
 
 describe('enqueueGroupedEmailFlush', () => {
-  it('triggers with a 5-minute delay and a workspace-scoped idempotency key', () => {
+  it('triggers with a 5-minute delay and a round-scoped idempotency key', () => {
     enqueueGroupedEmailFlush(payload)
 
     expect(flushGroupedEmail.trigger).toHaveBeenCalledTimes(1)
     const [triggeredPayload, opts] = (flushGroupedEmail.trigger as jest.Mock).mock.calls[0]
     expect(triggeredPayload).toEqual(payload)
     expect(opts).toMatchObject({
       delay: '5m',
-      idempotencyKey: 'ws_1:client_1:win_1',
+      idempotencyKey: 'ws_1:client_1:win_1:0',
       idempotencyKeyTTL: '10m',
     })
   })
+
+  it('waits longer and uses a distinct idempotency key on a retry round', () => {
+    enqueueGroupedEmailFlush({ ...payload, retryRound: 1 })
+
+    const [, opts] = (flushGroupedEmail.trigger as jest.Mock).mock.calls[0]
+    expect(opts).toMatchObject({ delay: '15m', idempotencyKey: 'ws_1:client_1:win_1:1' })
+  })
 })
 
 describe('flushGroupedEmailOnFailure', () => {
-  it('captures to Sentry with workspace and window tags', async () => {
+  it('re-enqueues the window instead of dropping buffered events', async () => {
+    await flushGroupedEmailOnFailure({ payload, error: new Error('db unreachable') })
+
+    expect(mockExecuteRaw).not.toHaveBeenCalled()
+    expect(mockCaptureException).not.toHaveBeenCalled()
+    const [triggeredPayload] = (flushGroupedEmail.trigger as jest.Mock).mock.calls[0]
+    expect(triggeredPayload).toEqual({ ...payload, retryRound: 1 })
+  })
+
+  it('captures to Sentry with workspace and window tags once rounds are exhausted', async () => {
     const error = new Error('terminal')
 
-    await flushGroupedEmailOnFailure({ payload, error })
+    await flushGroupedEmailOnFailure({ payload: { ...payload, retryRound: 2 }, error })
 
     expect(mockCaptureException).toHaveBeenCalledTimes(1)
     const [captured, opts] = mockCaptureException.mock.calls[0]
@@ -348,9 +364,10 @@ describe('flushGroupedEmailOnFailure', () => {
     })
   })
 
-  it('deletes all rows for the window so orphaned events do not accumulate', async () => {
-    await flushGroupedEmailOnFailure({ payload, error: new Error('terminal') })
+  it('deletes all rows for the window only after the last retry round', async () => {
+    await flushGroupedEmailOnFailure({ payload: { ...payload, retryRound: 2 }, error: new Error('terminal') })
 
     expect(mockExecuteRaw).toHaveBeenCalledTimes(1)
+    expect(flushGroupedEmail.trigger).not.toHaveBeenCalled()
   })
 })

src/jobs/notifications/flush-grouped-email.ts

@@ -18,6 +18,7 @@ import { sendGroupedEmail } from './send-grouped-email'
 export type FlushGroupedEmailPayload = {
   workspaceId: string
   windowKey: string
+  retryRound?: number
 }
 
 type WindowEvent = GroupedEmailEventInput & { individualEmail: NotificationRequestBody | null }
@@ -40,6 +41,10 @@ type IuRecipientGroup = {
 }
 
 const TASK_ID = 'flush-grouped-email'
+const FLUSH_DELAY = '5m'
+// A pooler blip outlives the in-run backoff, so a dead window gets re-enqueued instead of dropped.
+const RETRY_ROUND_DELAY = '15m'
+const MAX_RETRY_ROUNDS = 2
 
 const readUnsentWindowEvents = (db: ReturnType<typeof DBClient.getInstance>, windowKey: string) =>
   db.$queryRaw<BufferedRow[]>`
@@ -319,11 +324,24 @@ export const flushGroupedEmailRun = async (payload: FlushGroupedEmailPayload) =>
 }
 
 export const flushGroupedEmailOnFailure = async ({ payload, error }: { payload: unknown; error: unknown }) => {
-  const { workspaceId, windowKey } = payload as FlushGroupedEmailPayload
-  Sentry.captureException(error, { tags: { job: TASK_ID, workspaceId, windowKey } })
-  logger.error('flush-grouped-email: retries exhausted, cleaning up window', {
+  const { workspaceId, windowKey, retryRound = 0 } = payload as FlushGroupedEmailPayload
+
+  if (retryRound < MAX_RETRY_ROUNDS) {
+    logger.warn('flush-grouped-email: attempts exhausted, re-enqueueing window', {
+      workspaceId,
+      windowKey,
+      retryRound,
+      error: serializeError(error),
+    })
+    await enqueueGroupedEmailFlush({ workspaceId, windowKey, retryRound: retryRound + 1 })
+    return
+  }
+
+  Sentry.captureException(error, { tags: { job: TASK_ID, workspaceId, windowKey, retryRound: String(retryRound) } })
+  logger.error('flush-grouped-email: retry rounds exhausted, cleaning up window', {
     workspaceId,
     windowKey,
+    retryRound,
     error: serializeError(error),
   })
   const db = DBClient.getInstance()
@@ -341,7 +359,7 @@ export const flushGroupedEmailOnFailure = async ({ payload, error }: { payload:
 export const flushGroupedEmail = task({
   id: TASK_ID,
   queue: { concurrencyLimit: 5 },
-  retry: { maxAttempts: 3, factor: 2, minTimeoutInMs: 1_000, maxTimeoutInMs: 15_000, randomize: true },
+  retry: { maxAttempts: 5, factor: 2, minTimeoutInMs: 5_000, maxTimeoutInMs: 60_000, randomize: true },
   maxDuration: 60,
   run: flushGroupedEmailRun,
 })
@@ -350,7 +368,7 @@ tasks.onFailure(TASK_ID, flushGroupedEmailOnFailure)
 
 export const enqueueGroupedEmailFlush = (payload: FlushGroupedEmailPayload) =>
   flushGroupedEmail.trigger(payload, {
-    delay: '5m',
-    idempotencyKey: `${payload.workspaceId}:${payload.windowKey}`,
+    delay: payload.retryRound ? RETRY_ROUND_DELAY : FLUSH_DELAY,
+    idempotencyKey: `${payload.workspaceId}:${payload.windowKey}:${payload.retryRound ?? 0}`,
     idempotencyKeyTTL: '10m',
   })