From 090976e227ef8e6cd357eef03b4de3a5d4361a5d Mon Sep 17 00:00:00 2001 From: d-klotz Date: Tue, 12 May 2026 05:27:17 -0300 Subject: [PATCH] fix(core): gracefully discard queue messages when process not found MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a sync process is deleted mid-flight (e.g., completed or cancelled while pagination messages are still in the queue), loadIntegrationForProcess threw an unclassified error. Since the error has no statusCode, the queue worker retried it 4 times before sending it to DLQ — wasting Lambda invocations on a permanently unrecoverable condition. Now loadIntegrationForProcess returns null (matching loadIntegrationForWebhook pattern), and the QueueWorker discards the message with a warn log. Production evidence: 7 DLQ events/day for FETCH_PERSON_PAGE with integrationId: null, all processIds belonging to already-completed syncs. Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/core/handlers/backend-utils.js | 8 +- .../integration-defined-workers.test.js | 83 +++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/packages/core/handlers/backend-utils.js b/packages/core/handlers/backend-utils.js index 618d66527..8b03ee201 100644 --- a/packages/core/handlers/backend-utils.js +++ b/packages/core/handlers/backend-utils.js @@ -129,7 +129,7 @@ const loadIntegrationForProcess = async (processId, integrationClass) => { const process = await processRepository.findById(processId); if (!process) { - throw new Error(`Process not found: ${processId}`); + return null; } const instance = await getIntegrationInstance.execute( @@ -167,6 +167,12 @@ const createQueueWorker = (integrationClass) => { params.data.processId, integrationClass ); + if (!integrationInstance) { + console.warn( + `[${integrationName}] Process ${params.data.processId} not found. Discarding ${params.event} message.` + ); + return; + } console.log(`[QueueWorker] hydrated`, { ...logCtx, integrationStatus: integrationInstance?.status, diff --git a/packages/core/handlers/workers/integration-defined-workers.test.js b/packages/core/handlers/workers/integration-defined-workers.test.js index a10cc4bd7..ee20655d0 100644 --- a/packages/core/handlers/workers/integration-defined-workers.test.js +++ b/packages/core/handlers/workers/integration-defined-workers.test.js @@ -596,5 +596,88 @@ describe('Webhook Queue Worker', () => { await expect(worker.run(sqsEvent, {})).resolves.not.toThrow(); }); }); + + describe('Process not found handling (hydration by processId)', () => { + it('should discard message when process no longer exists', async () => { + const consoleSpy = jest.spyOn(console, 'warn').mockImplementation(); + + let mockedCreateQueueWorker; + jest.isolateModules(() => { + jest.doMock('../../integrations/repositories/process-repository-factory', () => ({ + createProcessRepository: () => ({ + findById: jest.fn().mockResolvedValue(null), + }), + })); + jest.doMock('../../integrations/repositories/integration-repository-factory', () => ({ + createIntegrationRepository: () => ({}), + })); + jest.doMock('../../modules/repositories/module-repository-factory', () => ({ + createModuleRepository: () => ({}), + })); + mockedCreateQueueWorker = require('../backend-utils').createQueueWorker; + }); + + const QueueWorker = mockedCreateQueueWorker(TestWebhookIntegration); + const worker = new QueueWorker(); + + const params = { + event: 'FETCH_PERSON_PAGE', + data: { + processId: 'deleted-process-12475', + personObjectType: 'Contact', + page: 3, + }, + }; + + const sqsEvent = { + Records: [{ messageId: 'msg-1', body: JSON.stringify(params) }], + }; + + const result = await worker.run(sqsEvent, {}); + + expect(result.batchItemFailures).toEqual([]); + expect(consoleSpy).toHaveBeenCalledWith( + expect.stringContaining('Process deleted-process-12475 not found') + ); + + consoleSpy.mockRestore(); + }); + + it('should still throw for other processId hydration errors', async () => { + let mockedCreateQueueWorker; + jest.isolateModules(() => { + jest.doMock('../../integrations/repositories/process-repository-factory', () => ({ + createProcessRepository: () => ({ + findById: jest.fn().mockRejectedValue(new Error('DB connection failed')), + }), + })); + jest.doMock('../../integrations/repositories/integration-repository-factory', () => ({ + createIntegrationRepository: () => ({}), + })); + jest.doMock('../../modules/repositories/module-repository-factory', () => ({ + createModuleRepository: () => ({}), + })); + mockedCreateQueueWorker = require('../backend-utils').createQueueWorker; + }); + + const QueueWorker = mockedCreateQueueWorker(TestWebhookIntegration); + const worker = new QueueWorker(); + + const params = { + event: 'FETCH_PERSON_PAGE', + data: { + processId: 'process-123', + personObjectType: 'Contact', + }, + }; + + const sqsEvent = { + Records: [{ messageId: 'msg-1', body: JSON.stringify(params) }], + }; + + const result = await worker.run(sqsEvent, {}); + expect(result.batchItemFailures).toEqual([{ itemIdentifier: 'msg-1' }]); + }); + }); });