OUT-4099 + OUT-4100: Assembly 429 backpressure + per-direction fan-out with per-portal fairness

src/features/sync/constant.ts

@@ -1,5 +1,8 @@
 export const MAX_FILES_LIMIT = 150
 
+// Items per batchTriggerAndWait call, kept within Trigger.dev's batch limit.
+export const BATCH_CHUNK_SIZE = 150
+
 export const DBX_URL_PATH = {
   fileUpload: '/files/upload',
   fileDownload: '/files/download',

src/features/webhook/dropbox/lib/webhook.service.ts

@@ -231,14 +231,22 @@ export class DropboxWebhook {
     }
 
     if (allChanges.length > 0) {
-      await handleChannelFileChanges.triggerAndWait({
+      const result = await handleChannelFileChanges.triggerAndWait({
         files: allChanges,
         channelSyncId,
         dbxRootPath,
         assemblyChannelId,
         user,
         connectionToken,
       })
+      // Don't advance the cursor if change processing failed, or these deltas move
+      // past the cursor and are never re-fetched. Throwing lets the run retry from
+      // the same cursor.
+      if (!result.ok) {
+        throw new Error(`handleChannelFileChanges failed for channel ${channelSyncId}`, {
+          cause: result.error,
+        })
+      }
     }
 
     await mapFilesService.updateChannelMapById(

src/lib/__tests__/fanOut.test.ts

@@ -0,0 +1,75 @@
+import { describe, expect, it, vi } from 'vitest'
+import { BATCH_CHUNK_SIZE } from '@/features/sync/constant'
+import { chunk, fanOutAndWait } from '@/lib/fanOut'
+
+describe('chunk', () => {
+  it('splits into batches of the given size', () => {
+    expect(chunk([1, 2, 3, 4, 5], 2)).toEqual([[1, 2], [3, 4], [5]])
+  })
+
+  it('returns an empty array for no items', () => {
+    expect(chunk([], 3)).toEqual([])
+  })
+})
+
+describe('fanOutAndWait', () => {
+  it('does not trigger the task for an empty list', async () => {
+    const task = { batchTriggerAndWait: vi.fn(() => Promise.resolve()) }
+
+    await fanOutAndWait(task, [] as { payload: string }[], 'portal-1')
+
+    expect(task.batchTriggerAndWait).not.toHaveBeenCalled()
+  })
+
+  it('attaches the concurrencyKey to every item', async () => {
+    type Item = { payload: string; options?: { concurrencyKey?: string } }
+    const task = { batchTriggerAndWait: vi.fn((_items: Item[]) => Promise.resolve()) }
+
+    await fanOutAndWait(task, [{ payload: 'a' }, { payload: 'b' }], 'portal-1')
+
+    expect(task.batchTriggerAndWait.mock.calls[0][0]).toEqual([
+      { payload: 'a', options: { concurrencyKey: 'portal-1' } },
+      { payload: 'b', options: { concurrencyKey: 'portal-1' } },
+    ])
+  })
+
+  it('merges the concurrencyKey into existing item options', async () => {
+    type Item = { payload: string; options?: Record<string, unknown> }
+    const task = { batchTriggerAndWait: vi.fn((_items: Item[]) => Promise.resolve()) }
+
+    await fanOutAndWait(task, [{ payload: 'a', options: { idempotencyKey: 'k1' } }], 'portal-1')
+
+    expect(task.batchTriggerAndWait.mock.calls[0][0]).toEqual([
+      { payload: 'a', options: { idempotencyKey: 'k1', concurrencyKey: 'portal-1' } },
+    ])
+  })
+
+  it('chunks large fan-outs by BATCH_CHUNK_SIZE', async () => {
+    const task = { batchTriggerAndWait: vi.fn(() => Promise.resolve()) }
+    const items = Array.from({ length: BATCH_CHUNK_SIZE * 2 + 1 }, (_, i) => ({ payload: i }))
+
+    await fanOutAndWait(task, items, 'portal-1')
+
+    expect(task.batchTriggerAndWait).toHaveBeenCalledTimes(3)
+  })
+
+  it('awaits batches sequentially — never two waits in flight (Trigger.dev bans parallel waits)', async () => {
+    let inFlight = 0
+    let maxInFlight = 0
+    const task = {
+      batchTriggerAndWait: vi.fn(async () => {
+        inFlight += 1
+        maxInFlight = Math.max(maxInFlight, inFlight)
+        await Promise.resolve()
+        await Promise.resolve()
+        inFlight -= 1
+      }),
+    }
+    const items = Array.from({ length: BATCH_CHUNK_SIZE * 2 + 1 }, (_, i) => ({ payload: i }))
+
+    await fanOutAndWait(task, items, 'portal-1')
+
+    expect(task.batchTriggerAndWait.mock.calls.length).toBeGreaterThan(1)
+    expect(maxInFlight).toBe(1)
+  })
+})

src/lib/__tests__/withRetry.test.ts

@@ -0,0 +1,70 @@
+import { DropboxResponseError } from 'dropbox'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { withRetry } from '@/lib/withRetry'
+import { sleep } from '@/utils/sleep'
+
+// Keep Sentry inert and sleep instant so retries don't wait in real time.
+vi.mock('@sentry/nextjs', () => ({
+  default: { withScope: (fn: (scope: unknown) => void) => fn({ addEventProcessor: vi.fn() }) },
+}))
+vi.mock('@/utils/sleep')
+
+// A Copilot error carries only a numeric status (no retry-after header).
+const copilot = (status: number) => Object.assign(new Error('copilot error'), { status })
+
+// Tiny pRetry backoff so tests stay fast; the 1s Copilot wait is mocked separately.
+const fast = { minTimeout: 1, maxTimeout: 2 }
+
+beforeEach(() => {
+  vi.mocked(sleep).mockReset()
+  vi.mocked(sleep).mockResolvedValue(undefined)
+})
+
+describe('withRetry — Copilot (Assembly) rate-limit backpressure', () => {
+  it('waits ~1s on a Copilot 429, then retries to success', async () => {
+    let calls = 0
+    const fn = vi.fn(() => {
+      calls += 1
+      return calls === 1 ? Promise.reject(copilot(429)) : Promise.resolve('ok')
+    })
+
+    const result = await withRetry(fn, [], fast)
+
+    expect(result).toBe('ok')
+    expect(fn).toHaveBeenCalledTimes(2)
+    expect(vi.mocked(sleep)).toHaveBeenCalledWith(1000)
+  })
+
+  it('does not retry a non-retryable Copilot status (404)', async () => {
+    const fn = vi.fn(() => Promise.reject(copilot(404)))
+
+    await expect(withRetry(fn, [], fast)).rejects.toThrow('copilot error')
+    expect(fn).toHaveBeenCalledTimes(1)
+  })
+
+  it('defaults to 3 retries (4 attempts) when no retries option is given', async () => {
+    const fn = vi.fn(() => Promise.reject(copilot(429)))
+
+    await expect(withRetry(fn, [], fast)).rejects.toThrow('copilot error')
+    expect(fn).toHaveBeenCalledTimes(4)
+  })
+
+  it('honors a raised retries option (retries: 6 → 7 attempts)', async () => {
+    const fn = vi.fn(() => Promise.reject(copilot(429)))
+
+    await expect(withRetry(fn, [], { ...fast, retries: 6 })).rejects.toThrow('copilot error')
+    expect(fn).toHaveBeenCalledTimes(7)
+    // Waited before each of the 6 retries, but not before the final give-up.
+    expect(vi.mocked(sleep)).toHaveBeenCalledTimes(6)
+  })
+
+  it('does not apply the fixed 1s Copilot wait to a Dropbox 429', async () => {
+    // Dropbox 429 must not trigger the Copilot 1s wait.
+    const dbxError = new DropboxResponseError(429, { get: () => null }, { error_summary: 'rate' })
+    const fn = vi.fn(() => Promise.reject(dbxError))
+
+    await expect(withRetry(fn, [], fast)).rejects.toBeInstanceOf(DropboxResponseError)
+    expect(vi.mocked(sleep)).not.toHaveBeenCalledWith(1000)
+    expect(fn.mock.calls.length).toBeGreaterThan(1)
+  })
+})

src/lib/copilot/CopilotAPI.ts

@@ -181,7 +181,8 @@ export class CopilotAPI {
   private wrapWithRetry<Args extends unknown[], R>(
     fn: (...args: Args) => Promise<R>,
   ): (...args: Args) => Promise<R> {
-    return (...args: Args): Promise<R> => withRetry(fn.bind(this), args)
+    // 6 retries so a 429 burst self-corrects here before task-level retry.
+    return (...args: Args): Promise<R> => withRetry(fn.bind(this), args, { retries: 6 })
   }
 
   // Methods wrapped with retry

src/lib/fanOut.ts

@@ -0,0 +1,28 @@
+import { BATCH_CHUNK_SIZE } from '@/features/sync/constant'
+
+// Split a large fan-out into batches; batchTriggerAndWait caps items per call.
+export const chunk = <T>(items: T[], size: number): T[][] =>
+  Array.from({ length: Math.ceil(items.length / size) }, (_, i) =>
+    items.slice(i * size, i * size + size),
+  )
+
+type BatchOptions = { concurrencyKey?: string; [key: string]: unknown }
+type BatchItem<TItem> = TItem & { options?: BatchOptions }
+type KeyedBatchTask<TItem> = {
+  batchTriggerAndWait: (items: BatchItem<TItem>[]) => Promise<unknown>
+}
+
+// Keyed per portal for fairness. Trigger.dev forbids parallel waits, so batches
+// are awaited one at a time (items within a batch still run concurrently).
+export const fanOutAndWait = async <TItem extends object>(
+  task: KeyedBatchTask<TItem>,
+  items: BatchItem<TItem>[],
+  concurrencyKey: string,
+): Promise<void> => {
+  if (!items.length) return
+  // Merge the key into any existing options instead of replacing them.
+  const keyed = items.map((item) => ({ ...item, options: { ...item.options, concurrencyKey } }))
+  for (const batch of chunk(keyed, BATCH_CHUNK_SIZE)) {
+    await task.batchTriggerAndWait(batch)
+  }
+}

src/lib/withRetry.ts

@@ -19,8 +19,9 @@ export const withRetry = async <Args extends unknown[], R>(
   fn: (...args: Args) => Promise<R>,
   args: Args,
   opts?: {
-    minTimeout: number
-    maxTimeout: number
+    minTimeout?: number
+    maxTimeout?: number
+    retries?: number
   },
 ): Promise<R> => {
   let isEventProcessorRegistered = false
@@ -64,17 +65,26 @@ export const withRetry = async <Args extends unknown[], R>(
     },
 
     {
-      retries: 3,
+      retries: opts?.retries ?? 3,
       minTimeout: opts?.minTimeout ?? 500,
       maxTimeout: opts?.maxTimeout ?? 2000,
       factor: 2, // Exponential factor for timeout delay. Tweak this if issues still persist
 
-      onFailedAttempt: (error: { error: unknown; attemptNumber: number; retriesLeft: number }) => {
-        if (error.error instanceof DropboxResponseError) {
-          if (!RETRYABLE_STATUS_CODES.has(error.error.status)) return
-        } else if (!RETRYABLE_STATUS_CODES.has((error.error as StatusableError).status)) {
-          return
+      onFailedAttempt: async (error: {
+        error: unknown
+        attemptNumber: number
+        retriesLeft: number
+      }) => {
+        const err = error.error
+        const isDropbox = err instanceof DropboxResponseError
+        const status = isDropbox ? err.status : (err as StatusableError).status
+        if (!RETRYABLE_STATUS_CODES.has(status)) return
+
+        // Copilot has no retry-after header; wait ~1s (its recovery) before an actual retry.
+        if (!isDropbox && status === httpStatus.TOO_MANY_REQUESTS && error.retriesLeft > 0) {
+          await sleep(1000)
         }
+
         console.warn(
           `CopilotAPI#withRetry | Attempt ${error.attemptNumber} failed. There are ${error.retriesLeft} retries left. Error:`,
           error,