diff --git a/src/features/sync/constant.ts b/src/features/sync/constant.ts index 11bce85..e5020c9 100644 --- a/src/features/sync/constant.ts +++ b/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', diff --git a/src/features/webhook/dropbox/lib/webhook.service.ts b/src/features/webhook/dropbox/lib/webhook.service.ts index 9ec5b57..ff3a93c 100644 --- a/src/features/webhook/dropbox/lib/webhook.service.ts +++ b/src/features/webhook/dropbox/lib/webhook.service.ts @@ -231,7 +231,7 @@ export class DropboxWebhook { } if (allChanges.length > 0) { - await handleChannelFileChanges.triggerAndWait({ + const result = await handleChannelFileChanges.triggerAndWait({ files: allChanges, channelSyncId, dbxRootPath, @@ -239,6 +239,14 @@ export class DropboxWebhook { 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( diff --git a/src/lib/__tests__/fanOut.test.ts b/src/lib/__tests__/fanOut.test.ts new file mode 100644 index 0000000..a74745e --- /dev/null +++ b/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 } + 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) + }) +}) diff --git a/src/lib/__tests__/withRetry.test.ts b/src/lib/__tests__/withRetry.test.ts new file mode 100644 index 0000000..19e6ced --- /dev/null +++ b/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) + }) +}) diff --git a/src/lib/copilot/CopilotAPI.ts b/src/lib/copilot/CopilotAPI.ts index 5a5626c..539443a 100644 --- a/src/lib/copilot/CopilotAPI.ts +++ b/src/lib/copilot/CopilotAPI.ts @@ -181,7 +181,8 @@ export class CopilotAPI { private wrapWithRetry( fn: (...args: Args) => Promise, ): (...args: Args) => Promise { - return (...args: Args): Promise => withRetry(fn.bind(this), args) + // 6 retries so a 429 burst self-corrects here before task-level retry. + return (...args: Args): Promise => withRetry(fn.bind(this), args, { retries: 6 }) } // Methods wrapped with retry diff --git a/src/lib/fanOut.ts b/src/lib/fanOut.ts new file mode 100644 index 0000000..38bb0e3 --- /dev/null +++ b/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 = (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 & { options?: BatchOptions } +type KeyedBatchTask = { + batchTriggerAndWait: (items: BatchItem[]) => Promise +} + +// 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 ( + task: KeyedBatchTask, + items: BatchItem[], + concurrencyKey: string, +): Promise => { + 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) + } +} diff --git a/src/lib/withRetry.ts b/src/lib/withRetry.ts index 2d25353..f707cd0 100644 --- a/src/lib/withRetry.ts +++ b/src/lib/withRetry.ts @@ -19,8 +19,9 @@ export const withRetry = async ( fn: (...args: Args) => Promise, args: Args, opts?: { - minTimeout: number - maxTimeout: number + minTimeout?: number + maxTimeout?: number + retries?: number }, ): Promise => { let isEventProcessorRegistered = false @@ -64,17 +65,26 @@ export const withRetry = async ( }, { - 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, diff --git a/src/trigger/processFileSync.ts b/src/trigger/processFileSync.ts index 3589ad9..ba3491a 100644 --- a/src/trigger/processFileSync.ts +++ b/src/trigger/processFileSync.ts @@ -22,6 +22,7 @@ import { CopilotAPI } from '@/lib/copilot/CopilotAPI' import type User from '@/lib/copilot/models/User.model' import { DropboxAuthClient } from '@/lib/dropbox/DropboxAuthClient' import { DropboxClient } from '@/lib/dropbox/DropboxClient' +import { fanOutAndWait } from '@/lib/fanOut' import { classifyDbxChanges } from '@/utils/classify-dbx-changes' import { withErrorLogging } from '@/utils/withErrorLogger' @@ -117,50 +118,59 @@ export const initiateDropboxToAssemblySync = task({ include_non_downloadable_files: false, }) - // 2. loop over the dropbox files - while (dbxFiles.result.entries.length) { - // refresh access token for every batch - await dbx.dbxAuthClient.refreshAccessToken(connectionToken.refreshToken) + // accumulate all pages, then fan out once + const allEntries: Awaited> = [] - const parsedDbxFiles = DropboxFileListFolderResultEntriesSchema.safeParse( - dbxFiles.result.entries, - ) - - if (!parsedDbxFiles.success) { - logger.error('Error parsing Dropbox files', { error: parsedDbxFiles.error }) - break - } - const parsedDbxEntries = parsedDbxFiles.data - - // check and filter out all the mapped files - const filteredEntries = await mapFilesService.checkAndFilterDbxFiles( - parsedDbxEntries, - dbxRootPath, - assemblyChannelId, - ) + // 2. loop over the dropbox files. Capture a mid-pagination failure so files + // already discovered are still dispatched below instead of being lost. + let listingError: unknown = null + try { + while (dbxFiles.result.entries.length) { + // refresh access token for every batch + await dbx.dbxAuthClient.refreshAccessToken(connectionToken.refreshToken) - if (filteredEntries.length) { - await syncDropboxFileToAssembly.batchTriggerAndWait(filteredEntries) - } + const parsedDbxFiles = DropboxFileListFolderResultEntriesSchema.safeParse( + dbxFiles.result.entries, + ) - if (!dbxFiles.result.has_more) { - // update channelSync with lastest cursor - await mapFilesService.updateChannelMap( - { - dbxCursor: dbxFiles.result.cursor, - }, - assemblyChannelId, + if (!parsedDbxFiles.success) { + logger.error('Error parsing Dropbox files', { error: parsedDbxFiles.error }) + // Throw (not break) so an incomplete listing doesn't get marked synced + // or advance the cursor; the catch still fans out the valid earlier pages. + throw parsedDbxFiles.error + } + const parsedDbxEntries = parsedDbxFiles.data + + // check and filter out all the mapped files + const filteredEntries = await mapFilesService.checkAndFilterDbxFiles( + parsedDbxEntries, dbxRootPath, + assemblyChannelId, ) - break - } - // continue pagination - dbxFiles = await dbxClient.filesListFolderContinue({ - cursor: dbxFiles.result.cursor, - }) + if (filteredEntries.length) allEntries.push(...filteredEntries) + + if (!dbxFiles.result.has_more) break + + // continue pagination + dbxFiles = await dbxClient.filesListFolderContinue({ + cursor: dbxFiles.result.cursor, + }) + } + } catch (error) { + listingError = error } + // 3. fan out all files, keyed per portal. Always dispatch what we discovered, + // even if a later page failed, so earlier pages aren't lost. + await fanOutAndWait(syncDropboxFileToAssembly, allEntries, user.portalId) + + // Don't mark synced or advance the cursor on an incomplete listing: advancing + // would skip the unlisted pages from the webhook delta baseline. Safe to bail — + // a later run re-lists from root and mapping-based filtering dedupes what synced. + if (listingError) throw listingError + + // mark synced only after the fan-out, so progress can't report 100% early await mapFilesService.updateChannelMap( { status: true, @@ -279,27 +289,41 @@ export const initiateAssemblyToDropboxSync = task({ const copilotApi = new CopilotAPI(payload.user.portalId) let files = await copilotApi.listFiles(payload.assemblyChannelId) - while (files.data.length) { - // refresh dropbox access token for every batch - await dbxAuth.refreshAccessToken(connectionToken.refreshToken) + // accumulate all pages, then fan out once + const allEntries: Awaited> = [] - // 2. check and filter out all the mapped files - const filteredEntries = await mapFilesService.checkAndFilterAssemblyFiles( - files.data, - dbxRootPath, - assemblyChannelId, - ) + // Capture a mid-pagination failure so files already discovered are still + // dispatched below instead of being lost. + let listingError: unknown = null + try { + while (files.data.length) { + // refresh dropbox access token for every batch + await dbxAuth.refreshAccessToken(connectionToken.refreshToken) - if (filteredEntries.length) { - await syncAssemblyFileToDropbox.batchTriggerAndWait(filteredEntries) - } + // 2. check and filter out all the mapped files + const filteredEntries = await mapFilesService.checkAndFilterAssemblyFiles( + files.data, + dbxRootPath, + assemblyChannelId, + ) - if (!files.nextToken) { - break - } + if (filteredEntries.length) allEntries.push(...filteredEntries) + + if (!files.nextToken) { + break + } - files = await copilotApi.listFiles(payload.assemblyChannelId, files.nextToken) + files = await copilotApi.listFiles(payload.assemblyChannelId, files.nextToken) + } + } catch (error) { + listingError = error } + + // fan out all files, keyed per portal. Always dispatch what we discovered, + // even if a later page failed, so earlier pages aren't lost. + await fanOutAndWait(syncAssemblyFileToDropbox, allEntries, user.portalId) + + if (listingError) throw listingError }, })