From df6fcf7552db5a9eeb3899c239b452e810b124bb Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Tue, 25 Aug 2026 18:22:47 +0545 Subject: [PATCH 1/4] fix(OUT-4099): recover from Assembly 429s in withRetry Assembly returns no retry-after header and recovers in ~1s. Previously a 429 fell through to blind exponential backoff and escalated to the 2-5 min task-level retry, parking files. Now a Copilot 429 waits ~1s (only before an actual retry, not the final give-up) and the Copilot path retries up to 6 times so a burst self-corrects inside withRetry. Dropbox callers keep their existing 3-retry behaviour via a new configurable `retries` option. Co-Authored-By: Claude Opus 4.8 --- src/lib/__tests__/withRetry.test.ts | 70 +++++++++++++++++++++++++++++ src/lib/copilot/CopilotAPI.ts | 3 +- src/lib/withRetry.ts | 26 +++++++---- 3 files changed, 90 insertions(+), 9 deletions(-) create mode 100644 src/lib/__tests__/withRetry.test.ts 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/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, From 2feaad193f6ae225a7393e6abff170c9ced63ea8 Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Tue, 25 Aug 2026 18:23:02 +0545 Subject: [PATCH 2/4] perf(OUT-4100): fan out files per direction, keyed per portal Replace the per-page batchTriggerAndWait convoy with accumulate-then-fan-out: each direction lists all pages first, then dispatches files in chunks (BATCH_CHUNK_SIZE, within Trigger.dev's 500-item batch limit). Chunks are awaited sequentially since Trigger.dev forbids parallel waits. Each file run is keyed by concurrencyKey=portalId so one busy portal can no longer starve others. The A->D-before-D->A ordering barrier is preserved (each direction awaits its own fan-out) and the channel is marked synced only after it. Co-Authored-By: Claude Opus 4.8 --- src/features/sync/constant.ts | 3 ++ src/lib/__tests__/fanOut.test.ts | 64 ++++++++++++++++++++++++++++++++ src/lib/fanOut.ts | 26 +++++++++++++ src/trigger/processFileSync.ts | 34 ++++++++--------- 4 files changed, 110 insertions(+), 17 deletions(-) create mode 100644 src/lib/__tests__/fanOut.test.ts create mode 100644 src/lib/fanOut.ts 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/lib/__tests__/fanOut.test.ts b/src/lib/__tests__/fanOut.test.ts new file mode 100644 index 0000000..484932d --- /dev/null +++ b/src/lib/__tests__/fanOut.test.ts @@ -0,0 +1,64 @@ +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('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/fanOut.ts b/src/lib/fanOut.ts new file mode 100644 index 0000000..7109c80 --- /dev/null +++ b/src/lib/fanOut.ts @@ -0,0 +1,26 @@ +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 BatchItemWithKey = TItem & { options?: { concurrencyKey?: string } } +type KeyedBatchTask = { + batchTriggerAndWait: (items: BatchItemWithKey[]) => 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: TItem[], + concurrencyKey: string, +): Promise => { + if (!items.length) return + const keyed = items.map((item) => ({ ...item, options: { concurrencyKey } })) + for (const batch of chunk(keyed, BATCH_CHUNK_SIZE)) { + await task.batchTriggerAndWait(batch) + } +} diff --git a/src/trigger/processFileSync.ts b/src/trigger/processFileSync.ts index 3589ad9..e2c8755 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,6 +118,9 @@ export const initiateDropboxToAssemblySync = task({ include_non_downloadable_files: false, }) + // accumulate all pages, then fan out once + const allEntries: Awaited> = [] + // 2. loop over the dropbox files while (dbxFiles.result.entries.length) { // refresh access token for every batch @@ -139,21 +143,9 @@ export const initiateDropboxToAssemblySync = task({ assemblyChannelId, ) - if (filteredEntries.length) { - await syncDropboxFileToAssembly.batchTriggerAndWait(filteredEntries) - } + if (filteredEntries.length) allEntries.push(...filteredEntries) - if (!dbxFiles.result.has_more) { - // update channelSync with lastest cursor - await mapFilesService.updateChannelMap( - { - dbxCursor: dbxFiles.result.cursor, - }, - assemblyChannelId, - dbxRootPath, - ) - break - } + if (!dbxFiles.result.has_more) break // continue pagination dbxFiles = await dbxClient.filesListFolderContinue({ @@ -161,6 +153,10 @@ export const initiateDropboxToAssemblySync = task({ }) } + // 3. fan out all files, keyed per portal + await fanOutAndWait(syncDropboxFileToAssembly, allEntries, user.portalId) + + // mark synced only after the fan-out, so progress can't report 100% early await mapFilesService.updateChannelMap( { status: true, @@ -279,6 +275,9 @@ export const initiateAssemblyToDropboxSync = task({ const copilotApi = new CopilotAPI(payload.user.portalId) let files = await copilotApi.listFiles(payload.assemblyChannelId) + // accumulate all pages, then fan out once + const allEntries: Awaited> = [] + while (files.data.length) { // refresh dropbox access token for every batch await dbxAuth.refreshAccessToken(connectionToken.refreshToken) @@ -290,9 +289,7 @@ export const initiateAssemblyToDropboxSync = task({ assemblyChannelId, ) - if (filteredEntries.length) { - await syncAssemblyFileToDropbox.batchTriggerAndWait(filteredEntries) - } + if (filteredEntries.length) allEntries.push(...filteredEntries) if (!files.nextToken) { break @@ -300,6 +297,9 @@ export const initiateAssemblyToDropboxSync = task({ files = await copilotApi.listFiles(payload.assemblyChannelId, files.nextToken) } + + // fan out all files, keyed per portal + await fanOutAndWait(syncAssemblyFileToDropbox, allEntries, user.portalId) }, }) From 99be660a2a8e1206aac2df7d2fa7ea9141e8ed30 Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Tue, 25 Aug 2026 19:07:21 +0545 Subject: [PATCH 3/4] =?UTF-8?q?fix(OUT-4100):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20keep=20pages=20on=20pagination=20failure,=20merge?= =?UTF-8?q?=20options?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - initiate* now wrap pagination in try/catch: if a later page fails, files already discovered are still fanned out, then the error is rethrown so the channel isn't marked synced and the cursor isn't advanced on an incomplete listing (a later run re-lists from root; mapping-based filtering dedupes). - fanOutAndWait merges concurrencyKey into existing item options instead of replacing them, so future scheduling options aren't silently dropped. Co-Authored-By: Claude Opus 4.8 --- src/lib/__tests__/fanOut.test.ts | 11 +++ src/lib/fanOut.ts | 10 +-- src/trigger/processFileSync.ts | 112 ++++++++++++++++++------------- 3 files changed, 84 insertions(+), 49 deletions(-) diff --git a/src/lib/__tests__/fanOut.test.ts b/src/lib/__tests__/fanOut.test.ts index 484932d..a74745e 100644 --- a/src/lib/__tests__/fanOut.test.ts +++ b/src/lib/__tests__/fanOut.test.ts @@ -33,6 +33,17 @@ describe('fanOutAndWait', () => { ]) }) + 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 })) diff --git a/src/lib/fanOut.ts b/src/lib/fanOut.ts index 7109c80..38bb0e3 100644 --- a/src/lib/fanOut.ts +++ b/src/lib/fanOut.ts @@ -6,20 +6,22 @@ export const chunk = (items: T[], size: number): T[][] => items.slice(i * size, i * size + size), ) -type BatchItemWithKey = TItem & { options?: { concurrencyKey?: string } } +type BatchOptions = { concurrencyKey?: string; [key: string]: unknown } +type BatchItem = TItem & { options?: BatchOptions } type KeyedBatchTask = { - batchTriggerAndWait: (items: BatchItemWithKey[]) => Promise + 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: TItem[], + items: BatchItem[], concurrencyKey: string, ): Promise => { if (!items.length) return - const keyed = items.map((item) => ({ ...item, options: { concurrencyKey } })) + // 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/trigger/processFileSync.ts b/src/trigger/processFileSync.ts index e2c8755..b8fd603 100644 --- a/src/trigger/processFileSync.ts +++ b/src/trigger/processFileSync.ts @@ -121,41 +121,53 @@ export const initiateDropboxToAssemblySync = task({ // accumulate all pages, then fan out once const allEntries: Awaited> = [] - // 2. loop over the dropbox files - while (dbxFiles.result.entries.length) { - // refresh access token for every batch - await dbx.dbxAuthClient.refreshAccessToken(connectionToken.refreshToken) - - const parsedDbxFiles = DropboxFileListFolderResultEntriesSchema.safeParse( - dbxFiles.result.entries, - ) - - if (!parsedDbxFiles.success) { - logger.error('Error parsing Dropbox files', { error: parsedDbxFiles.error }) - break + // 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) + + 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, + ) + + if (filteredEntries.length) allEntries.push(...filteredEntries) + + if (!dbxFiles.result.has_more) break + + // continue pagination + dbxFiles = await dbxClient.filesListFolderContinue({ + cursor: dbxFiles.result.cursor, + }) } - const parsedDbxEntries = parsedDbxFiles.data - - // check and filter out all the mapped files - const filteredEntries = await mapFilesService.checkAndFilterDbxFiles( - parsedDbxEntries, - dbxRootPath, - assemblyChannelId, - ) - - 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 + // 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( { @@ -278,28 +290,38 @@ export const initiateAssemblyToDropboxSync = task({ // accumulate all pages, then fan out once const allEntries: Awaited> = [] - while (files.data.length) { - // refresh dropbox access token for every batch - await dbxAuth.refreshAccessToken(connectionToken.refreshToken) + // 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) + + // 2. check and filter out all the mapped files + const filteredEntries = await mapFilesService.checkAndFilterAssemblyFiles( + files.data, + dbxRootPath, + assemblyChannelId, + ) - // 2. check and filter out all the mapped files - const filteredEntries = await mapFilesService.checkAndFilterAssemblyFiles( - files.data, - dbxRootPath, - assemblyChannelId, - ) + if (filteredEntries.length) allEntries.push(...filteredEntries) - if (filteredEntries.length) allEntries.push(...filteredEntries) + if (!files.nextToken) { + break + } - 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 + // 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 }, }) From 475734ec9781593a82c3c55883cbc7e250bc4e69 Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Tue, 25 Aug 2026 19:33:07 +0545 Subject: [PATCH 4/4] fix(OUT-4100): don't advance sync state past a failed listing Co-Authored-By: Claude Opus 4.8 --- src/features/webhook/dropbox/lib/webhook.service.ts | 10 +++++++++- src/trigger/processFileSync.ts | 4 +++- 2 files changed, 12 insertions(+), 2 deletions(-) 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/trigger/processFileSync.ts b/src/trigger/processFileSync.ts index b8fd603..ba3491a 100644 --- a/src/trigger/processFileSync.ts +++ b/src/trigger/processFileSync.ts @@ -135,7 +135,9 @@ export const initiateDropboxToAssemblySync = task({ if (!parsedDbxFiles.success) { logger.error('Error parsing Dropbox files', { error: parsedDbxFiles.error }) - break + // 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