fix(newsletters): fence resend recovery state

Author: Bill Leoutsakos

apps/sim/app/api/superuser/newsletters/runs/[id]/export.csv/route.ts

@@ -5,6 +5,7 @@ import { exportNewsletterRunCsvContract } from '@/lib/api/contracts/newsletters'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { validateNewsletterSuperuser } from '@/lib/newsletters/auth'
+import { isNewsletterResendError } from '@/lib/newsletters/resend'
 import { createNewsletterCsvExport } from '@/lib/newsletters/runs'
 
 const logger = createLogger('NewsletterCsvExportAPI')
@@ -58,7 +59,7 @@ export const GET = withRouteHandler(
       if (/Finalize/i.test(message)) {
         return NextResponse.json({ error: message }, { status: 400 })
       }
-      if (/RESEND_API_KEY|Resend .*list|Resend request/i.test(message)) {
+      if (isNewsletterResendError(error)) {
         return NextResponse.json({ error: message }, { status: 503 })
       }
       logger.error('Failed to export newsletter CSV', { error: message })

apps/sim/app/api/superuser/newsletters/runs/[id]/finalize/route.test.ts

@@ -0,0 +1,63 @@
+/**
+ * @vitest-environment node
+ */
+import { createMockRequest } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockFinalizeNewsletterRun, mockValidateNewsletterSuperuser } = vi.hoisted(() => ({
+  mockFinalizeNewsletterRun: vi.fn(),
+  mockValidateNewsletterSuperuser: vi.fn(),
+}))
+
+vi.mock('@/lib/newsletters/auth', () => ({
+  validateNewsletterSuperuser: mockValidateNewsletterSuperuser,
+}))
+
+vi.mock('@/lib/newsletters/runs', () => ({
+  finalizeNewsletterRun: mockFinalizeNewsletterRun,
+}))
+
+import { NewsletterResendError } from '@/lib/newsletters/resend'
+import { POST } from '@/app/api/superuser/newsletters/runs/[id]/finalize/route'
+
+function callRoute() {
+  const request = createMockRequest(
+    'POST',
+    undefined,
+    {},
+    'http://localhost:3000/api/superuser/newsletters/runs/run-1/finalize'
+  )
+  return POST(request, { params: Promise.resolve({ id: 'run-1' }) })
+}
+
+describe('newsletter run finalization', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockValidateNewsletterSuperuser.mockResolvedValue({
+      success: true,
+      userId: 'admin-1',
+    })
+  })
+
+  it.each([
+    'Resend suppression pagination returned no cursor',
+    'Resend contact pagination returned no cursor',
+    'Resend contact property pagination returned no cursor',
+  ])('maps a Resend service failure to 503: %s', async (message) => {
+    mockFinalizeNewsletterRun.mockRejectedValueOnce(new NewsletterResendError(message))
+
+    const response = await callRoute()
+
+    expect(response.status).toBe(503)
+    await expect(response.json()).resolves.toEqual({ error: message })
+  })
+
+  it('does not classify an unrelated error by message text', async () => {
+    mockFinalizeNewsletterRun.mockRejectedValueOnce(new Error('Resend text from unrelated code'))
+
+    const response = await callRoute()
+
+    expect(response.status).toBe(500)
+    await expect(response.json()).resolves.toEqual({ error: 'Internal server error' })
+  })
+})

apps/sim/app/api/superuser/newsletters/runs/[id]/finalize/route.ts

@@ -5,6 +5,7 @@ import { finalizeNewsletterRunContract } from '@/lib/api/contracts/newsletters'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { validateNewsletterSuperuser } from '@/lib/newsletters/auth'
+import { isNewsletterResendError } from '@/lib/newsletters/resend'
 import { finalizeNewsletterRun } from '@/lib/newsletters/runs'
 
 const logger = createLogger('NewsletterFinalizeAPI')
@@ -35,7 +36,7 @@ export const POST = withRouteHandler(
       if (/already in progress/i.test(message)) {
         return NextResponse.json({ error: message }, { status: 409 })
       }
-      if (/RESEND_API_KEY|Resend .*list|Resend request/i.test(message)) {
+      if (isNewsletterResendError(error)) {
         return NextResponse.json({ error: message }, { status: 503 })
       }
       logger.error('Failed to finalize newsletter run', { error: message })

apps/sim/lib/api/contracts/newsletters.ts

@@ -114,7 +114,7 @@ export const newsletterRunResponseSchema = z.object({
 
 export const pushNewsletterRunResponseSchema = z.object({
   run: newsletterRunSchema,
-  jobId: z.string(),
+  jobId: z.string().nullable(),
 })
 
 export const newsletterJobResponseSchema = z.object({

apps/sim/lib/newsletters/push-resend.test.ts

@@ -58,6 +58,7 @@ vi.mock('@/lib/newsletters/runs', () => ({
   updateRecipientSyncStatus: mocks.updateRecipient,
 }))
 
+import { pushNewsletterRunResponseSchema } from '@/lib/api/contracts/newsletters'
 import { enqueueNewsletterResendSync, runNewsletterResendSync } from '@/lib/newsletters/push-resend'
 
 const run = {
@@ -86,6 +87,7 @@ describe('newsletter Resend queueing', () => {
     })
     mocks.queueEnqueue.mockResolvedValue('trigger-run-123')
     mocks.setJob.mockResolvedValue({ ...run, resendSyncJobId: 'trigger-run-123' })
+    mocks.setSegment.mockResolvedValue(run)
     mocks.isAsyncJobEnqueueError.mockReturnValue(false)
   })
 
@@ -238,6 +240,55 @@ describe('newsletter Resend queueing', () => {
     }
   )
 
+  it('does not enqueue a pushed run without a stored job id', async () => {
+    const pushedRun = { ...run, status: 'pushed' }
+    mocks.claimAttempt.mockResolvedValue({
+      attempt: 2,
+      jobId: null,
+      run: pushedRun,
+      shouldEnqueue: false,
+    })
+
+    const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+    expect(mocks.getJobQueue).not.toHaveBeenCalled()
+    expect(mocks.queueEnqueue).not.toHaveBeenCalled()
+    expect(result).toEqual({ run: pushedRun, jobId: null })
+    expect(pushNewsletterRunResponseSchema.shape.jobId.parse(result.jobId)).toBeNull()
+  })
+
+  it.each([
+    ['completed', { id: 'trigger-run-existing', status: 'completed' }],
+    ['failed', { id: 'trigger-run-existing', status: 'failed', error: 'worker failed' }],
+    ['processing', { id: 'trigger-run-existing', status: 'processing' }],
+  ])(
+    'does not enqueue when reconciliation of a %s job refreshes to pushed',
+    async (_providerState, providerJob) => {
+      const pushedRun = { ...run, status: 'pushed' }
+      mocks.claimAttempt
+        .mockResolvedValueOnce({
+          attempt: 2,
+          jobId: 'trigger-run-existing',
+          run,
+          shouldEnqueue: false,
+        })
+        .mockResolvedValueOnce({
+          attempt: 2,
+          jobId: null,
+          run: pushedRun,
+          shouldEnqueue: false,
+        })
+      mocks.queueGetJob.mockResolvedValue(providerJob)
+
+      const result = await enqueueNewsletterResendSync('run-1', 'admin-1')
+
+      expect(mocks.markFailed).toHaveBeenCalledWith('run-1', 2, expect.any(Error))
+      expect(mocks.queueEnqueue).not.toHaveBeenCalled()
+      expect(mocks.setJob).not.toHaveBeenCalled()
+      expect(result).toEqual({ run: pushedRun, jobId: null })
+    }
+  )
+
   it('re-enqueues when a stored Trigger.dev run no longer exists', async () => {
     mocks.claimAttempt
       .mockResolvedValueOnce({
@@ -342,6 +393,78 @@ describe('newsletter Resend queueing', () => {
     expect(mocks.markFailed).not.toHaveBeenCalled()
   })
 
+  it('stops when the authoritative post-reset read is already pushed', async () => {
+    mocks.requireAttempt
+      .mockResolvedValueOnce(run)
+      .mockResolvedValueOnce({ ...run, status: 'pushed' })
+
+    await runNewsletterResendSync({
+      runId: 'run-1',
+      attempt: 2,
+      requestedById: 'admin-1',
+    })
+
+    expect(mocks.createSegment).not.toHaveBeenCalled()
+    expect(mocks.setSegment).not.toHaveBeenCalled()
+    expect(mocks.ensureProperties).not.toHaveBeenCalled()
+    expect(mocks.markFailed).not.toHaveBeenCalled()
+  })
+
+  it('uses the segment from the authoritative post-reset read', async () => {
+    const staleRun = {
+      ...run,
+      resendSegmentId: null,
+      resendSegmentName: null,
+    }
+    const currentRun = {
+      ...run,
+      resendSegmentId: 'segment-current',
+      resendSegmentName: 'Current segment',
+    }
+    mocks.requireAttempt.mockResolvedValueOnce(staleRun).mockResolvedValueOnce(currentRun)
+    mocks.setSegment.mockResolvedValue(currentRun)
+    mocks.getExcludedEmails.mockResolvedValue(new Set())
+    mocks.getPendingRecipients.mockResolvedValue([])
+    mocks.countByStatus.mockResolvedValue({})
+
+    await runNewsletterResendSync({
+      runId: 'run-1',
+      attempt: 2,
+      requestedById: 'admin-1',
+    })
+
+    expect(mocks.createSegment).not.toHaveBeenCalled()
+    expect(mocks.setSegment).toHaveBeenCalledWith('run-1', 2, 'segment-current', 'Current segment')
+    expect(mocks.markPushed).toHaveBeenCalledWith('run-1', 2, 'segment-current', 'Current segment')
+  })
+
+  it('stops when another worker pushes while a segment is being created', async () => {
+    const runWithoutSegment = {
+      ...run,
+      resendSegmentId: null,
+      resendSegmentName: null,
+    }
+    mocks.requireAttempt.mockResolvedValue(runWithoutSegment)
+    mocks.createSegment.mockResolvedValue({ id: 'segment-new', name: 'New segment' })
+    mocks.setSegment.mockResolvedValue({
+      ...run,
+      status: 'pushed',
+      resendSegmentId: 'segment-winner',
+      resendSegmentName: 'Winner segment',
+    })
+
+    await runNewsletterResendSync({
+      runId: 'run-1',
+      attempt: 2,
+      requestedById: 'admin-1',
+    })
+
+    expect(mocks.setSegment).toHaveBeenCalledWith('run-1', 2, 'segment-new', 'New segment')
+    expect(mocks.ensureProperties).not.toHaveBeenCalled()
+    expect(mocks.markPushed).not.toHaveBeenCalled()
+    expect(mocks.markFailed).not.toHaveBeenCalled()
+  })
+
   it('does not fail the newsletter attempt when its database claim is aborted', async () => {
     const controller = new AbortController()
     controller.abort('claim lost')

apps/sim/lib/newsletters/push-resend.ts

@@ -74,17 +74,32 @@ export async function runNewsletterResendSync(
       throw new Error('Newsletter run must be finalized before pushing to Resend')
     }
     await resetFailedNewsletterRecipients(runId, attempt)
-    await requireNewsletterRunAttempt(runId, attempt)
+    const currentRun = await requireNewsletterRunAttempt(runId, attempt)
+    if (currentRun.status === 'pushed') return
+    if (currentRun.status !== 'pushing' && currentRun.status !== 'failed') {
+      throw new Error('Newsletter run is not eligible to continue its Resend sync')
+    }
 
     signal?.throwIfAborted()
-    const segment =
-      run.resendSegmentId && run.resendSegmentName
-        ? { id: run.resendSegmentId, name: run.resendSegmentName }
-        : await createNewsletterSegment(segmentNameForRun(run.name), { signal })
+    const segmentCandidate =
+      currentRun.resendSegmentId && currentRun.resendSegmentName
+        ? { id: currentRun.resendSegmentId, name: currentRun.resendSegmentName }
+        : await createNewsletterSegment(segmentNameForRun(currentRun.name), { signal })
 
     signal?.throwIfAborted()
-    if (!run.resendSegmentId) {
-      await setNewsletterRunResendSegment(runId, attempt, segment.id, segment.name)
+    const segmentRun = await setNewsletterRunResendSegment(
+      runId,
+      attempt,
+      segmentCandidate.id,
+      segmentCandidate.name
+    )
+    if (segmentRun.status === 'pushed') return
+    if (!segmentRun.resendSegmentId || !segmentRun.resendSegmentName) {
+      throw new Error('Newsletter Resend segment tracking is incomplete')
+    }
+    const segment = {
+      id: segmentRun.resendSegmentId,
+      name: segmentRun.resendSegmentName,
     }
 
     signal?.throwIfAborted()
@@ -189,17 +204,21 @@ export async function runNewsletterResendSync(
 
 export async function enqueueNewsletterResendSync(runId: string, requestedById: string) {
   let claim = await claimNewsletterRunResendAttempt(runId)
+  if (claim.run.status === 'pushed') {
+    return { run: claim.run, jobId: claim.jobId }
+  }
+
   const queue = await getJobQueue()
   const backendType = getAsyncBackendType()
   if (!claim.shouldEnqueue && claim.jobId) {
-    if (claim.run.status === 'pushed') {
-      return { run: claim.run, jobId: claim.jobId }
-    }
     const persistedJob = await queue.getJob(claim.jobId)
     if (persistedJob?.status === JOB_STATUS.COMPLETED) {
       const error = new Error('Newsletter sync job completed without finalizing the newsletter run')
       await markNewsletterRunPushFailed(runId, claim.attempt, error)
       claim = await claimNewsletterRunResendAttempt(runId)
+      if (claim.run.status === 'pushed') {
+        return { run: claim.run, jobId: claim.jobId }
+      }
     } else if (backendType !== 'database') {
       if (
         persistedJob?.status === JOB_STATUS.PENDING ||
@@ -212,10 +231,16 @@ export async function enqueueNewsletterResendSync(runId: string, requestedById:
       )
       await markNewsletterRunPushFailed(runId, claim.attempt, error)
       claim = await claimNewsletterRunResendAttempt(runId)
+      if (claim.run.status === 'pushed') {
+        return { run: claim.run, jobId: claim.jobId }
+      }
     } else if (persistedJob?.status === JOB_STATUS.FAILED) {
       const error = new Error(persistedJob.error ?? 'Newsletter sync job failed')
       await markNewsletterRunPushFailed(runId, claim.attempt, error)
       claim = await claimNewsletterRunResendAttempt(runId)
+      if (claim.run.status === 'pushed') {
+        return { run: claim.run, jobId: claim.jobId }
+      }
     }
   }
 
@@ -254,5 +279,5 @@ export async function enqueueNewsletterResendSync(runId: string, requestedById:
       { cause: error }
     )
   }
-  return { run: updatedRun, jobId }
+  return { run: updatedRun, jobId: updatedRun.resendSyncJobId ?? jobId }
 }

apps/sim/lib/newsletters/resend.test.ts

@@ -22,6 +22,8 @@ import {
   ensureNewsletterContactProperties,
   getResendExcludedEmails,
   getResendSuppressedEmails,
+  isNewsletterResendError,
+  NewsletterResendError,
 } from '@/lib/newsletters/resend'
 
 function jsonResponse(body: unknown, status = 200) {
@@ -50,6 +52,11 @@ describe('newsletter Resend service', () => {
     )
   })
 
+  it('classifies only typed Resend service failures', () => {
+    expect(isNewsletterResendError(new NewsletterResendError('provider unavailable'))).toBe(true)
+    expect(isNewsletterResendError(new Error('Resend text from unrelated code'))).toBe(false)
+  })
+
   it('does not make a Resend request when already aborted', async () => {
     const controller = new AbortController()
     controller.abort('cancelled')
@@ -60,6 +67,21 @@ describe('newsletter Resend service', () => {
     expect(fetchMock).not.toHaveBeenCalled()
   })
 
+  it('preserves cancellation while reading a Resend error response', async () => {
+    const controller = new AbortController()
+    const body = new ReadableStream({
+      pull(streamController) {
+        controller.abort('cancelled while reading')
+        streamController.error(new Error('body read failed'))
+      },
+    })
+    fetchMock.mockResolvedValueOnce(new Response(body, { status: 400 }))
+
+    await expect(createNewsletterSegment('Segment 1', { signal: controller.signal })).rejects.toBe(
+      'cancelled while reading'
+    )
+  })
+
   it('does not retry after cancellation during backoff', async () => {
     const controller = new AbortController()
     fetchMock.mockResolvedValueOnce(jsonResponse({ message: 'retry' }, 429))