From 4463d47440c30329bc082bfa204b2872d9712b7c Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Wed, 26 Aug 2026 20:50:42 +0545 Subject: [PATCH 1/2] perf(OUT-4101): process webhook changes per channel, not globally Co-Authored-By: Claude Opus 4.8 --- .../webhook/dropbox/lib/webhook.service.ts | 19 +++++++++------- src/trigger/processFileSync.ts | 22 ++++++++++++------- 2 files changed, 25 insertions(+), 16 deletions(-) diff --git a/src/features/webhook/dropbox/lib/webhook.service.ts b/src/features/webhook/dropbox/lib/webhook.service.ts index ff3a93c..ae5dea1 100644 --- a/src/features/webhook/dropbox/lib/webhook.service.ts +++ b/src/features/webhook/dropbox/lib/webhook.service.ts @@ -231,14 +231,17 @@ export class DropboxWebhook { } if (allChanges.length > 0) { - const result = await handleChannelFileChanges.triggerAndWait({ - files: allChanges, - channelSyncId, - dbxRootPath, - assemblyChannelId, - user, - connectionToken, - }) + const result = await handleChannelFileChanges.triggerAndWait( + { + files: allChanges, + channelSyncId, + dbxRootPath, + assemblyChannelId, + user, + connectionToken, + }, + { concurrencyKey: channelSyncId }, + ) // 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. diff --git a/src/trigger/processFileSync.ts b/src/trigger/processFileSync.ts index ba3491a..ebec996 100644 --- a/src/trigger/processFileSync.ts +++ b/src/trigger/processFileSync.ts @@ -232,13 +232,15 @@ export const handleChannelFileChanges = task({ * After partial unique index: First delete and then create the files. This ensures that constraints are not violated. * This section should handle file rename, folder rename cases */ + // Keyed per channel so different channels run in parallel instead of the whole + // system going one-at-a-time; delete-before-create ordering is preserved by + // awaiting each step in turn. if (deleted.length) - await deleteDropboxFileInAssembly.batchTriggerAndWait(deleted.map(toPayload)) - if (created.length) await syncDropboxFileToAssembly.batchTriggerAndWait(created.map(toPayload)) - - for (const entry of contentUpdated) { - await updateDropboxFileInAssembly.triggerAndWait({ opts, entry }) - } + await fanOutAndWait(deleteDropboxFileInAssembly, deleted.map(toPayload), channelSyncId) + if (created.length) + await fanOutAndWait(syncDropboxFileToAssembly, created.map(toPayload), channelSyncId) + if (contentUpdated.length) + await fanOutAndWait(updateDropboxFileInAssembly, contentUpdated.map(toPayload), channelSyncId) }, }) @@ -265,8 +267,12 @@ export const updateDropboxFileInAssembly = task({ }, retry: RETRY_CONFIG, run: async (payload: DropboxToAssemblySyncFilesPayload) => { - await deleteDropboxFileInAssembly.triggerAndWait(payload) - await syncDropboxFileToAssembly.trigger(payload) + const { channelSyncId } = payload.opts + // Await the recreate (not fire-and-forget) so the mapping row is committed before + // this update reports done — otherwise a follow-up webhook for the same file can + // read stale state and dispatch a second concurrent sync of the same id. + await deleteDropboxFileInAssembly.triggerAndWait(payload, { concurrencyKey: channelSyncId }) + await syncDropboxFileToAssembly.triggerAndWait(payload, { concurrencyKey: channelSyncId }) }, }) From 1985413baae2c26b6aa901258e93ac5e29cce8ff Mon Sep 17 00:00:00 2001 From: SandipBajracharya Date: Wed, 26 Aug 2026 20:50:44 +0545 Subject: [PATCH 2/2] fix(OUT-4101): dedupe delta entries by id per bucket Co-Authored-By: Claude Opus 4.8 --- .../__tests__/classify-dbx-changes.test.ts | 50 +++++++++++++++++++ src/utils/classify-dbx-changes.ts | 18 ++++++- 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/src/utils/__tests__/classify-dbx-changes.test.ts b/src/utils/__tests__/classify-dbx-changes.test.ts index 162fbc0..a107776 100644 --- a/src/utils/__tests__/classify-dbx-changes.test.ts +++ b/src/utils/__tests__/classify-dbx-changes.test.ts @@ -133,4 +133,54 @@ describe('classifyDbxChanges', () => { expect(ids(result.created)).toEqual(['add']) expect(ids(result.contentUpdated)).toEqual(['upd']) }) + + it('dedupes duplicate ids within a bucket, keeping the last occurrence', () => { + const entries = [ + makeEntry({ id: 'dup', content_hash: 'new-1' }), + makeEntry({ id: 'dup', content_hash: 'new-2' }), + ] + const rows = [makeRow('dup', 'old')] + + const result = classifyDbxChanges(entries, rows) + + expect(ids(result.contentUpdated)).toEqual(['dup']) + expect(result.contentUpdated[0].content_hash).toBe('new-2') + }) + + it('keeps a rename pair intact (same id in delete and create buckets)', () => { + const entries = [ + makeEntry({ id: 'x', '.tag': 'deleted' }), + makeEntry({ id: 'x', '.tag': 'file', path_display: '/folder/renamed.txt' }), + ] + const rows = [makeRow('x')] + + const result = classifyDbxChanges(entries, rows) + + expect(ids(result.deleted)).toEqual(['x']) + expect(ids(result.created)).toEqual(['x']) + }) + + it('dedupes duplicate ids within the created bucket', () => { + const entries = [ + makeEntry({ id: 'new', path_display: '/folder/a.txt' }), + makeEntry({ id: 'new', path_display: '/folder/b.txt' }), + ] + + const result = classifyDbxChanges(entries, []) + + expect(ids(result.created)).toEqual(['new']) + expect(result.created[0].path_display).toBe('/folder/b.txt') + }) + + it('dedupes duplicate ids within the deleted bucket', () => { + const entries = [ + makeEntry({ id: 'gone', '.tag': 'deleted' }), + makeEntry({ id: 'gone', '.tag': 'deleted' }), + ] + const rows = [makeRow('gone')] + + const result = classifyDbxChanges(entries, rows) + + expect(ids(result.deleted)).toEqual(['gone']) + }) }) diff --git a/src/utils/classify-dbx-changes.ts b/src/utils/classify-dbx-changes.ts index 041a82f..788fc77 100644 --- a/src/utils/classify-dbx-changes.ts +++ b/src/utils/classify-dbx-changes.ts @@ -10,6 +10,18 @@ export type DbxChangeClassification = { contentUpdated: DropboxFileListFolderSingleEntry[] } +// Keep only the last entry per id, preserving order. Guards against a delta carrying +// duplicate ids in one bucket, which would otherwise fan out concurrent same-file syncs +// and double-create or trip the partial unique index. Applied per bucket so a rename's +// delete+create pair (same id, different buckets) is left intact. +const dedupeByIdKeepLast = ( + entries: DropboxFileListFolderSingleEntry[], +): DropboxFileListFolderSingleEntry[] => { + const byId = new Map() + for (const entry of entries) byId.set(entry.id, entry) + return [...byId.values()] +} + /** * Splits Dropbox delta entries into deletes, creates, and content updates by comparing * them against the already-mapped rows. The caller must process `deleted` before `created` @@ -42,5 +54,9 @@ export const classifyDbxChanges = ( return !!existing?.contentHash && existing.contentHash !== entry.content_hash }) - return { deleted, created, contentUpdated } + return { + deleted: dedupeByIdKeepLast(deleted), + created: dedupeByIdKeepLast(created), + contentUpdated: dedupeByIdKeepLast(contentUpdated), + } }