fix(newsletters): propagate sync cancellation

Bill Leoutsakos · Bill Leoutsakos · commit f8c79bf2c009 · 2026-07-29T18:20:11.000-07:00
diff --git a/apps/sim/background/newsletter-resend-sync.test.ts b/apps/sim/background/newsletter-resend-sync.test.ts
@@ -0,0 +1,40 @@
+/**
+ * @vitest-environment node
+ */
+import { expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  runSync: vi.fn(),
+  task: vi.fn((definition) => definition),
+}))
+
+vi.mock('@trigger.dev/sdk', () => ({
+  task: mocks.task,
+}))
+
+vi.mock('@/lib/newsletters/push-resend', () => ({
+  NEWSLETTER_RESEND_SYNC_CONCURRENCY_LIMIT: 1,
+  NEWSLETTER_RESEND_SYNC_MAX_ATTEMPTS: 3,
+  runNewsletterResendSync: mocks.runSync,
+}))
+
+import type { NewsletterResendSyncPayload } from '@/lib/newsletters/push-resend'
+import { newsletterResendSyncTask } from '@/background/newsletter-resend-sync'
+
+interface NewsletterTaskDefinition {
+  run: (payload: NewsletterResendSyncPayload, context: { signal: AbortSignal }) => Promise<unknown>
+}
+
+it('forwards the Trigger.dev cancellation signal to the sync service', async () => {
+  const payload = {
+    runId: 'run-1',
+    attempt: 1,
+    requestedById: 'admin-1',
+  }
+  const controller = new AbortController()
+  const definition = newsletterResendSyncTask as unknown as NewsletterTaskDefinition
+
+  await definition.run(payload, { signal: controller.signal })
+
+  expect(mocks.runSync).toHaveBeenCalledWith(payload, controller.signal)
+})
diff --git a/apps/sim/background/newsletter-resend-sync.ts b/apps/sim/background/newsletter-resend-sync.ts
@@ -15,5 +15,6 @@ export const newsletterResendSyncTask = task({
   queue: {
     concurrencyLimit: NEWSLETTER_RESEND_SYNC_CONCURRENCY_LIMIT,
   },
-  run: async (payload: NewsletterResendSyncPayload) => runNewsletterResendSync(payload),
+  run: async (payload: NewsletterResendSyncPayload, { signal }) =>
+    runNewsletterResendSync(payload, signal),
 })
diff --git a/apps/sim/lib/newsletters/push-resend.test.ts b/apps/sim/lib/newsletters/push-resend.test.ts
@@ -6,6 +6,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 const mocks = vi.hoisted(() => ({
   claimAttempt: vi.fn(),
   countByStatus: vi.fn(),
+  createContact: vi.fn(),
   createSegment: vi.fn(),
   ensureProperties: vi.fn(),
   getAsyncBackendType: vi.fn(),
@@ -39,7 +40,7 @@ vi.mock('@/lib/core/async-jobs', () => ({
 
 vi.mock('@/lib/newsletters/resend', () => ({
   createNewsletterSegment: mocks.createSegment,
-  createOrSegmentNewsletterContact: vi.fn(),
+  createOrSegmentNewsletterContact: mocks.createContact,
   ensureNewsletterContactProperties: mocks.ensureProperties,
   getResendExcludedEmails: mocks.getExcludedEmails,
 }))
@@ -387,6 +388,46 @@ describe('newsletter Resend queueing', () => {
     expect(mocks.markFailed).not.toHaveBeenCalled()
   })
 
+  it('forwards cancellation to every Resend operation', async () => {
+    const controller = new AbortController()
+    const recipient = {
+      currentUserEligible: true,
+      email: 'user@example.com',
+      name: 'Ada Lovelace',
+      simUnsubscribed: false,
+      snapshotVersion: 1,
+      userId: 'user-1',
+    }
+    mocks.requireAttempt.mockResolvedValue({
+      ...run,
+      resendSegmentId: null,
+      resendSegmentName: null,
+    })
+    mocks.createSegment.mockResolvedValue({ id: 'segment-new', name: 'Segment new' })
+    mocks.getExcludedEmails.mockResolvedValue(new Set())
+    mocks.getPendingRecipients.mockResolvedValueOnce([recipient]).mockResolvedValueOnce([])
+    mocks.createContact.mockResolvedValue({ status: 'created', contactId: 'contact-1' })
+    mocks.countByStatus.mockResolvedValue({})
+
+    await runNewsletterResendSync(
+      {
+        runId: 'run-1',
+        attempt: 2,
+        requestedById: 'admin-1',
+      },
+      controller.signal
+    )
+
+    expect(mocks.createSegment).toHaveBeenCalledWith(expect.any(String), {
+      signal: controller.signal,
+    })
+    expect(mocks.ensureProperties).toHaveBeenCalledWith({ signal: controller.signal })
+    expect(mocks.getExcludedEmails).toHaveBeenCalledWith({ signal: controller.signal })
+    expect(mocks.createContact).toHaveBeenCalledWith(
+      expect.objectContaining({ signal: controller.signal })
+    )
+  })
+
   it('does not mark a run pushed after ownership is lost during status aggregation', async () => {
     const controller = new AbortController()
     mocks.requireAttempt.mockResolvedValue(run)
diff --git a/apps/sim/lib/newsletters/push-resend.ts b/apps/sim/lib/newsletters/push-resend.ts
@@ -55,6 +55,7 @@ async function runWithConcurrency<T>(
   for (let index = 0; index < items.length; index += limit) {
     signal?.throwIfAborted()
     await Promise.all(items.slice(index, index + limit).map(worker))
+    signal?.throwIfAborted()
   }
 }
 
@@ -79,17 +80,17 @@ export async function runNewsletterResendSync(
     const segment =
       run.resendSegmentId && run.resendSegmentName
         ? { id: run.resendSegmentId, name: run.resendSegmentName }
-        : await createNewsletterSegment(segmentNameForRun(run.name))
+        : await createNewsletterSegment(segmentNameForRun(run.name), { signal })
 
     signal?.throwIfAborted()
     if (!run.resendSegmentId) {
       await setNewsletterRunResendSegment(runId, attempt, segment.id, segment.name)
     }
 
     signal?.throwIfAborted()
-    await ensureNewsletterContactProperties()
+    await ensureNewsletterContactProperties({ signal })
     signal?.throwIfAborted()
-    const suppressedEmails = await getResendExcludedEmails()
+    const suppressedEmails = await getResendExcludedEmails({ signal })
     let processed = 0
 
     while (true) {
@@ -128,6 +129,7 @@ export async function runNewsletterResendSync(
               userId: recipient.userId,
               runId,
               segmentId: segment.id,
+              signal,
             })
             signal?.throwIfAborted()
             await updateRecipientSyncStatus(
diff --git a/apps/sim/lib/newsletters/resend.test.ts b/apps/sim/lib/newsletters/resend.test.ts
@@ -3,15 +3,21 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { fetchMock } = vi.hoisted(() => ({
+const { fetchMock, sleepMock } = vi.hoisted(() => ({
   fetchMock: vi.fn(),
+  sleepMock: vi.fn(),
 }))
 
 vi.mock('@/lib/core/config/env', () => ({
   env: { RESEND_API_KEY: 're_test' },
 }))
 
+vi.mock('@sim/utils/helpers', () => ({
+  sleep: sleepMock,
+}))
+
 import {
+  createNewsletterSegment,
   createOrSegmentNewsletterContact,
   ensureNewsletterContactProperties,
   getResendExcludedEmails,
@@ -29,6 +35,42 @@ describe('newsletter Resend service', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     vi.stubGlobal('fetch', fetchMock)
+    sleepMock.mockResolvedValue(undefined)
+  })
+
+  it('forwards cancellation to Resend requests', async () => {
+    const controller = new AbortController()
+    fetchMock.mockResolvedValueOnce(jsonResponse({ id: 'segment-1', name: 'Segment 1' }, 201))
+
+    await createNewsletterSegment('Segment 1', { signal: controller.signal })
+
+    expect(fetchMock).toHaveBeenCalledWith(
+      'https://api.resend.com/segments',
+      expect.objectContaining({ signal: controller.signal })
+    )
+  })
+
+  it('does not make a Resend request when already aborted', async () => {
+    const controller = new AbortController()
+    controller.abort('cancelled')
+
+    await expect(createNewsletterSegment('Segment 1', { signal: controller.signal })).rejects.toBe(
+      'cancelled'
+    )
+    expect(fetchMock).not.toHaveBeenCalled()
+  })
+
+  it('does not retry after cancellation during backoff', async () => {
+    const controller = new AbortController()
+    fetchMock.mockResolvedValueOnce(jsonResponse({ message: 'retry' }, 429))
+    sleepMock.mockImplementationOnce(async () => {
+      controller.abort('cancelled')
+    })
+
+    await expect(createNewsletterSegment('Segment 1', { signal: controller.signal })).rejects.toBe(
+      'cancelled'
+    )
+    expect(fetchMock).toHaveBeenCalledTimes(1)
   })
 
   it('creates missing newsletter contact properties', async () => {
@@ -124,6 +166,28 @@ describe('newsletter Resend service', () => {
     )
   })
 
+  it('does not add segment membership after cancellation', async () => {
+    const controller = new AbortController()
+    fetchMock
+      .mockResolvedValueOnce(jsonResponse({ message: 'Contact already exists' }, 409))
+      .mockImplementationOnce(async () => {
+        controller.abort('cancelled')
+        return jsonResponse({ id: 'contact-1' })
+      })
+
+    await expect(
+      createOrSegmentNewsletterContact({
+        email: 'user@example.com',
+        name: 'Ada Lovelace',
+        userId: 'user-1',
+        runId: 'run-1',
+        segmentId: 'segment-1',
+        signal: controller.signal,
+      })
+    ).rejects.toBe('cancelled')
+    expect(fetchMock).toHaveBeenCalledTimes(2)
+  })
+
   it('normalizes suppressed email addresses', async () => {
     fetchMock.mockResolvedValueOnce(
       jsonResponse({
diff --git a/apps/sim/lib/newsletters/resend.ts b/apps/sim/lib/newsletters/resend.ts
@@ -42,6 +42,7 @@ interface ResendRequestOptions {
   body?: Record<string, unknown>
   required?: boolean
   maxResponseBytes?: number
+  signal?: AbortSignal
 }
 
 const resendSuppressionListSchema = z.object({
@@ -88,13 +89,15 @@ async function resendRequest<T>(path: string, options: ResendRequestOptions = {}
   if (!key) return undefined as T
 
   for (let attempt = 0; attempt < RESEND_MAX_ATTEMPTS; attempt++) {
+    options.signal?.throwIfAborted()
     const response = await fetch(`${RESEND_API_BASE}${path}`, {
       method: options.method ?? 'GET',
       headers: {
         Authorization: `Bearer ${key}`,
         'Content-Type': 'application/json',
       },
       body: options.body ? JSON.stringify(options.body) : undefined,
+      signal: options.signal,
     })
 
     if (response.ok) {
@@ -108,6 +111,7 @@ async function resendRequest<T>(path: string, options: ResendRequestOptions = {}
     if (response.status === 429 || response.status >= 500) {
       const retryAfterMs = parseRetryAfter(response.headers.get('retry-after'))
       await sleep(backoffWithJitter(attempt + 1, retryAfterMs, { baseMs: 250, maxMs: 5000 }))
+      options.signal?.throwIfAborted()
       continue
     }
 
@@ -117,7 +121,10 @@ async function resendRequest<T>(path: string, options: ResendRequestOptions = {}
   throw new Error('Resend request failed after retries')
 }
 
-export async function getResendSuppressedEmails(options?: { required?: boolean }) {
+export async function getResendSuppressedEmails(options?: {
+  required?: boolean
+  signal?: AbortSignal
+}) {
   const emails = new Set<string>()
   let after: string | null = null
   let hasMore = false
@@ -127,6 +134,7 @@ export async function getResendSuppressedEmails(options?: { required?: boolean }
     if (after) query.set('after', after)
     const rawResponse = await resendRequest<unknown>(`/suppressions?${query.toString()}`, {
       required: options?.required ?? true,
+      signal: options?.signal,
     })
     if (rawResponse === undefined) return emails
     const parsedResponse = resendSuppressionListSchema.safeParse(rawResponse)
@@ -152,8 +160,13 @@ export async function getResendSuppressedEmails(options?: { required?: boolean }
   return emails
 }
 
-export async function getResendExcludedEmails(): Promise<Set<string>> {
-  const excludedEmails = await getResendSuppressedEmails({ required: true })
+export async function getResendExcludedEmails(options?: {
+  signal?: AbortSignal
+}): Promise<Set<string>> {
+  const excludedEmails = await getResendSuppressedEmails({
+    required: true,
+    signal: options?.signal,
+  })
   let after: string | null = null
   let hasMore = false
 
@@ -162,6 +175,7 @@ export async function getResendExcludedEmails(): Promise<Set<string>> {
     if (after) query.set('after', after)
     const rawContacts = await resendRequest<unknown>(`/contacts?${query.toString()}`, {
       required: true,
+      signal: options?.signal,
     })
     const parsedContacts = resendContactListSchema.safeParse(rawContacts)
     if (!parsedContacts.success) {
@@ -188,14 +202,20 @@ export async function getResendExcludedEmails(): Promise<Set<string>> {
   return excludedEmails
 }
 
-export async function createNewsletterSegment(name: string): Promise<ResendSegmentResponse> {
+export async function createNewsletterSegment(
+  name: string,
+  options?: { signal?: AbortSignal }
+): Promise<ResendSegmentResponse> {
   return resendRequest<ResendSegmentResponse>('/segments', {
     method: 'POST',
     body: { name },
+    signal: options?.signal,
   })
 }
 
-export async function ensureNewsletterContactProperties(): Promise<void> {
+export async function ensureNewsletterContactProperties(options?: {
+  signal?: AbortSignal
+}): Promise<void> {
   const existingKeys = new Set<string>()
   let after: string | null = null
   let hasMore = false
@@ -204,7 +224,8 @@ export async function ensureNewsletterContactProperties(): Promise<void> {
     const query = new URLSearchParams({ limit: String(RESEND_CONTACT_PROPERTY_PAGE_LIMIT) })
     if (after) query.set('after', after)
     const response = await resendRequest<ResendContactPropertyListResponse>(
-      `/contact-properties?${query.toString()}`
+      `/contact-properties?${query.toString()}`,
+      { signal: options?.signal }
     )
 
     for (const property of response.data ?? []) {
@@ -228,6 +249,7 @@ export async function ensureNewsletterContactProperties(): Promise<void> {
         await resendRequest('/contact-properties', {
           method: 'POST',
           body: { key, type: 'string' },
+          signal: options?.signal,
         })
       } catch (error) {
         const message = getErrorMessage(error)
@@ -252,6 +274,7 @@ export async function createOrSegmentNewsletterContact(input: {
   userId: string | null
   runId: string
   segmentId: string
+  signal?: AbortSignal
 }): Promise<{ status: 'created' | 'updated' | 'segment_added'; contactId?: string }> {
   const email = normalizeEmail(input.email)
   const name = splitName(input.name)
@@ -269,6 +292,7 @@ export async function createOrSegmentNewsletterContact(input: {
         },
         segments: [{ id: input.segmentId }],
       },
+      signal: input.signal,
     })
     return { status: 'created', contactId: contact.id }
   } catch (error) {
@@ -286,10 +310,12 @@ export async function createOrSegmentNewsletterContact(input: {
           newsletter_run_id: input.runId,
         },
       },
+      signal: input.signal,
     }
   )
   await resendRequest(`/contacts/${encodeURIComponent(email)}/segments/${input.segmentId}`, {
     method: 'POST',
+    signal: input.signal,
   })
   return { status: 'updated', contactId: contact.id }
 }
