diff --git a/packages/@n8n/config/src/configs/workflows.config.ts b/packages/@n8n/config/src/configs/workflows.config.ts index fef46d31541..7aa30d5892b 100644 --- a/packages/@n8n/config/src/configs/workflows.config.ts +++ b/packages/@n8n/config/src/configs/workflows.config.ts @@ -7,6 +7,9 @@ import { positiveIntSchema } from '../schemas'; const callerPolicySchema = z.enum(['any', 'none', 'workflowsFromAList', 'workflowsFromSameOwner']); type CallerPolicy = z.infer; +// Bounded so lease-derived timeouts stay far below Node's max timer delay (~24.8 days). +const outboxLeaseSecondsSchema = positiveIntSchema.max(Time.days.toSeconds); + @Config export class WorkflowsConfig { /** Default name suggested when creating a new workflow. */ @@ -34,8 +37,9 @@ export class WorkflowsConfig { publicationOutboxPollIntervalMs: number = 15 * Time.seconds.toMilliseconds; /** Seconds after which an `in_progress` workflow publication outbox record - * is considered stale (its leader likely died) and may be reclaimed by a poll cycle. */ - @Env('N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS') + * is considered stale (its leader likely died) and may be reclaimed by a poll cycle. + * Must be at most one day. */ + @Env('N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS', outboxLeaseSecondsSchema) publicationOutboxLeaseSeconds: number = 2 * Time.minutes.toSeconds; /** Number of workflow publication outbox records the leader processes in parallel per drain. */ diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index 91ca2570ca0..08deb571145 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -983,6 +983,24 @@ describe('GlobalConfig', () => { const config = Container.get(GlobalConfig); expect(config.workflows.workflowPublicationConcurrency).toBe(3); }); + + it('should reject an outbox lease above one day and fall back to the default', () => { + process.env = { N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS: `${100 * 24 * 60 * 60}` }; + const config = Container.get(GlobalConfig); + expect(config.workflows.publicationOutboxLeaseSeconds).toBe(120); + expect(consoleWarnMock).toHaveBeenCalledWith( + expect.stringContaining('N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS'), + ); + }); + + it('should reject an outbox lease of 0 and fall back to the default', () => { + process.env = { N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS: '0' }; + const config = Container.get(GlobalConfig); + expect(config.workflows.publicationOutboxLeaseSeconds).toBe(120); + expect(consoleWarnMock).toHaveBeenCalledWith( + expect.stringContaining('N8N_WORKFLOW_PUBLICATION_OUTBOX_LEASE_SECONDS'), + ); + }); }); it('should clamp password min length to valid range', () => { diff --git a/packages/@n8n/utils/src/sleep.test.ts b/packages/@n8n/utils/src/sleep.test.ts index 95da1bab86c..71a7e079996 100644 --- a/packages/@n8n/utils/src/sleep.test.ts +++ b/packages/@n8n/utils/src/sleep.test.ts @@ -30,24 +30,33 @@ describe('sleep', () => { expect(onResolve).toHaveBeenCalled(); }); - it('should reject immediately if abort signal is already aborted', async () => { + it('should reject with the abort reason if abort signal is already aborted', async () => { const abortController = new AbortController(); - abortController.abort(); + const reason = new Error('deadline'); + abortController.abort(reason); - await expect(sleep(1000, abortController.signal)).rejects.toThrow('Aborted'); + await expect(sleep(1000, abortController.signal)).rejects.toBe(reason); }); - it('should reject when abort signal is triggered during sleep', async () => { + it('should reject with the abort reason when abort signal is triggered during sleep', async () => { const abortController = new AbortController(); + const reason = new Error('deadline'); const onSettled = vi.fn(); const promise = sleep(1000, abortController.signal).catch(onSettled); await vi.advanceTimersByTimeAsync(50); - abortController.abort(); + abortController.abort(reason); await promise; - expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ message: 'Aborted' })); + expect(onSettled).toHaveBeenCalledWith(reason); + }); + + it('should reject with the default abort reason when none is given', async () => { + const abortController = new AbortController(); + abortController.abort(); + + await expect(sleep(1000, abortController.signal)).rejects.toThrow('This operation was aborted'); }); it('should clean up timeout when aborted during sleep', async () => { @@ -59,10 +68,22 @@ describe('sleep', () => { await vi.advanceTimersByTimeAsync(50); abortController.abort(); - await expect(sleepPromise).rejects.toThrow('Aborted'); + await expect(sleepPromise).rejects.toThrow('This operation was aborted'); expect(clearTimeoutSpy).toHaveBeenCalled(); clearTimeoutSpy.mockRestore(); }); + + it('should remove the abort listener when the sleep completes', async () => { + const abortController = new AbortController(); + const removeListenerSpy = vi.spyOn(abortController.signal, 'removeEventListener'); + + const sleepPromise = sleep(100, abortController.signal); + + await vi.advanceTimersByTimeAsync(100); + await sleepPromise; + + expect(removeListenerSpy).toHaveBeenCalledWith('abort', expect.any(Function)); + }); }); diff --git a/packages/@n8n/utils/src/sleep.ts b/packages/@n8n/utils/src/sleep.ts index 3a7f7c4bc42..db568211254 100644 --- a/packages/@n8n/utils/src/sleep.ts +++ b/packages/@n8n/utils/src/sleep.ts @@ -1,25 +1,29 @@ +import { ensureError } from './errors/ensure-error'; + async function sleepWithAbort(ms: number, abortSignal: AbortSignal): Promise { return await new Promise((resolve, reject) => { if (abortSignal.aborted) { - reject(new Error('Aborted')); + reject(ensureError(abortSignal.reason)); return; } - const timeout = setTimeout(resolve, ms); + const onAbort = () => { + clearTimeout(timeout); + reject(ensureError(abortSignal.reason)); + }; - abortSignal.addEventListener( - 'abort', - () => { - clearTimeout(timeout); - reject(new Error('Aborted')); - }, - { once: true }, - ); + const timeout = setTimeout(() => { + abortSignal.removeEventListener('abort', onAbort); + resolve(); + }, ms); + + abortSignal.addEventListener('abort', onAbort, { once: true }); }); } /** - * Resolves after `ms` milliseconds, or rejects early if `abortSignal` is aborted. + * Resolves after `ms` milliseconds, or rejects with the signal's abort reason + * if `abortSignal` fires first. */ export async function sleep(ms: number, abortSignal?: AbortSignal): Promise { if (!abortSignal) { diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts index 6f5a80e3655..19d1d8f9708 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts @@ -93,6 +93,8 @@ describe('WorkflowPublicationApplier', () => { const newVersion = makeVersion('v-2'); const oldVersion = makeVersion('v-1'); + const abort = { signal: new AbortController().signal, onDetached: vi.fn() }; + /** Drives the trigger diff: first call returns old triggers, second returns new. */ function setTriggerSets(oldTriggers: INode[], newTriggers: INode[]) { workflowTriggerActivator.getEnabledTriggerNodes @@ -121,7 +123,7 @@ describe('WorkflowPublicationApplier', () => { test('skips with workflow-not-found when the workflow is gone', async () => { workflowRepository.findOneBy.mockResolvedValue(null); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'skipped', reason: 'workflow-not-found' }); expect(workflowTriggerActivator.getEnabledTriggerNodes).not.toHaveBeenCalled(); @@ -145,7 +147,7 @@ describe('WorkflowPublicationApplier', () => { triggerNode('b'), ]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'unpublished' }); expect(workflowTriggerActivator.getEnabledTriggerNodes).toHaveBeenCalledWith(oldVersion); @@ -153,6 +155,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), oldVersion, new Set(['a', 'b']), + abort, ); expect(workflowPublishedVersionRepository.removePublishedVersion).toHaveBeenCalledWith( 'wf-1', @@ -168,7 +171,7 @@ describe('WorkflowPublicationApplier', () => { ); workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'unpublished' }); expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled(); @@ -184,7 +187,7 @@ describe('WorkflowPublicationApplier', () => { // can leave rows behind after the mapping was removed. workflowPublishedVersionRepository.findOne.mockResolvedValue(makePublishedVersion(null)); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'unpublished' }); expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled(); @@ -201,7 +204,7 @@ describe('WorkflowPublicationApplier', () => { workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([triggerNode('a')]); workflowTriggerActivator.deactivate.mockRejectedValue(new Error('teardown boom')); - await expect(applier.apply(makeRecord())).rejects.toThrow('teardown boom'); + await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('teardown boom'); expect(workflowPublishedVersionRepository.removePublishedVersion).not.toHaveBeenCalled(); }); }); @@ -209,7 +212,7 @@ describe('WorkflowPublicationApplier', () => { test('returns version-missing when the published version history row is gone', async () => { workflowHistoryRepository.findOneBy.mockResolvedValue(null); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'version-missing' }); expect(workflowTriggerActivator.getEnabledTriggerNodes).not.toHaveBeenCalled(); @@ -220,7 +223,7 @@ describe('WorkflowPublicationApplier', () => { const trigger = triggerNode('a'); setTriggerSets([trigger], [{ ...trigger }]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -255,7 +258,7 @@ describe('WorkflowPublicationApplier', () => { failures: [], }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -269,7 +272,7 @@ describe('WorkflowPublicationApplier', () => { test('registers only added triggers', async () => { setTriggerSets([triggerNode('a')], [triggerNode('a'), triggerNode('b')]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -284,6 +287,7 @@ describe('WorkflowPublicationApplier', () => { newVersion, new Set(['b']), 'update', + abort, ); expect(workflowPublishedVersionRepository.setPublishedVersion).toHaveBeenCalledWith( 'wf-1', @@ -300,26 +304,28 @@ describe('WorkflowPublicationApplier', () => { ] as const)('reason %s activates with mode %s', async (reason, expectedMode) => { setTriggerSets([], [triggerNode('a')]); - await applier.apply(makeRecord({ reason })); + await applier.apply(makeRecord({ reason }), abort); expect(workflowTriggerActivator.activate).toHaveBeenCalledWith( expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), expectedMode, + abort, ); }); test('a record without a reason (pre-migration row) activates with update', async () => { setTriggerSets([], [triggerNode('a')]); - await applier.apply(makeRecord({ reason: undefined })); + await applier.apply(makeRecord({ reason: undefined }), abort); expect(workflowTriggerActivator.activate).toHaveBeenCalledWith( expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), 'update', + abort, ); }); }); @@ -333,7 +339,7 @@ describe('WorkflowPublicationApplier', () => { new Set(['a']), ); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -351,6 +357,7 @@ describe('WorkflowPublicationApplier', () => { newVersion, new Set(['a']), 'update', + abort, ); }); @@ -359,7 +366,7 @@ describe('WorkflowPublicationApplier', () => { setTriggerSets([trigger], [{ ...trigger }]); workflowTriggerActivator.getUnregisteredNonWebhookTriggerNodeIds.mockReturnValue(new Set()); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -383,7 +390,7 @@ describe('WorkflowPublicationApplier', () => { setTriggerSets([trigger], [{ ...trigger }]); workflowTriggerActivator.getNodesWithUnregisteredWebhooks.mockResolvedValue(new Set(['a'])); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -401,6 +408,7 @@ describe('WorkflowPublicationApplier', () => { newVersion, new Set(['a']), 'update', + abort, ); }); @@ -409,7 +417,7 @@ describe('WorkflowPublicationApplier', () => { setTriggerSets([triggerNode('a')], [triggerNode('a'), triggerNode('b')]); workflowTriggerActivator.getNodesWithUnregisteredWebhooks.mockResolvedValue(new Set(['c'])); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -423,13 +431,14 @@ describe('WorkflowPublicationApplier', () => { newVersion, new Set(['b', 'c']), 'update', + abort, ); }); test('deregisters only removed triggers and refreshes the trigger count', async () => { setTriggerSets([triggerNode('a'), triggerNode('b')], [triggerNode('a')]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -441,6 +450,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), oldVersion, new Set(['b']), + abort, ); expect(workflowTriggerActivator.activate).not.toHaveBeenCalled(); expect(workflowTriggerActivator.updateTriggerCount).toHaveBeenCalledWith( @@ -473,7 +483,7 @@ describe('WorkflowPublicationApplier', () => { return { activated: ['a'], failures: [] }; }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -485,12 +495,14 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), oldVersion, new Set(['a']), + abort, ); expect(workflowTriggerActivator.activate).toHaveBeenCalledWith( expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), 'update', + abort, ); // The cache is invalidated before the version is advanced and repopulated // straight after, so the empty window never serves a stale version, all @@ -505,7 +517,7 @@ describe('WorkflowPublicationApplier', () => { // A teardown failure happens before the version advances, so it bubbles up // to the consumer (which turns it into a failed result) rather than leaving // a half-applied publication marked completed. - await expect(applier.apply(makeRecord())).rejects.toThrow('teardown failed'); + await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('teardown failed'); expect(workflowPublishedVersionRepository.setPublishedVersion).not.toHaveBeenCalled(); expect(workflowTriggerActivator.activate).not.toHaveBeenCalled(); @@ -515,7 +527,7 @@ describe('WorkflowPublicationApplier', () => { setTriggerSets([triggerNode('a')], [triggerNode('a'), triggerNode('b')]); workflowTriggerActivator.activate.mockRejectedValue(new Error('registration failed')); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'failed', @@ -535,7 +547,7 @@ describe('WorkflowPublicationApplier', () => { failures: [{ nodeId: 'b', nodeName: 'b', error }], }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'partial', @@ -566,7 +578,7 @@ describe('WorkflowPublicationApplier', () => { failures: [{ nodeId: 'b', nodeName: 'b', error }], }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'partial', @@ -595,7 +607,7 @@ describe('WorkflowPublicationApplier', () => { failures: [{ nodeId: 'b', nodeName: 'b', error }], }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); // A single failure passes its error through, preserving the type. expect(result).toEqual({ @@ -627,7 +639,7 @@ describe('WorkflowPublicationApplier', () => { ]; workflowTriggerActivator.activate.mockResolvedValue({ activated: [], failures }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); // Nothing is running, so the publication failed; the combined error names both nodes. expect(result).toEqual({ @@ -664,7 +676,7 @@ describe('WorkflowPublicationApplier', () => { failures: [{ nodeId: 'b', nodeName: 'b', error }], }); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'partial', @@ -689,7 +701,7 @@ describe('WorkflowPublicationApplier', () => { workflowPublishedVersionRepository.findOne.mockResolvedValue(null); setTriggerSets([], [triggerNode('a')]); - const result = await applier.apply(makeRecord()); + const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'completed', @@ -703,6 +715,54 @@ describe('WorkflowPublicationApplier', () => { newVersion, new Set(['a']), 'update', + abort, ); }); + + describe('abort', () => { + function abortedContext() { + const controller = new AbortController(); + controller.abort(new Error('deadline')); + return { signal: controller.signal, onDetached: vi.fn() }; + } + + test('a publish aborted before teardown neither deactivates nor advances the version', async () => { + setTriggerSets([triggerNode('a')], []); + + await expect(applier.apply(makeRecord(), abortedContext())).rejects.toThrow('deadline'); + + expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled(); + expect(workflowPublishedVersionRepository.setPublishedVersion).not.toHaveBeenCalled(); + }); + + test('a publish aborted after teardown advances the version but fails before activation', async () => { + setTriggerSets([triggerNode('a')], [triggerNode('b')]); + const controller = new AbortController(); + workflowTriggerActivator.deactivate.mockImplementation(async () => { + controller.abort(new Error('deadline')); + }); + + const result = await applier.apply(makeRecord(), { + signal: controller.signal, + onDetached: vi.fn(), + }); + + expect(result).toEqual({ + type: 'failed', + error: expect.objectContaining({ message: 'deadline' }), + }); + expect(workflowPublishedVersionRepository.setPublishedVersion).toHaveBeenCalled(); + expect(workflowTriggerActivator.activate).not.toHaveBeenCalled(); + }); + + test('an unpublish aborted before teardown leaves the published-version mapping in place', async () => { + workflowRepository.findOneBy.mockResolvedValue(makeWorkflow({ activeVersionId: null })); + workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([triggerNode('a')]); + + await expect(applier.apply(makeRecord(), abortedContext())).rejects.toThrow('deadline'); + + expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled(); + expect(workflowPublishedVersionRepository.removePublishedVersion).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts index 62c6d7090b0..0dcb27044de 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts @@ -21,17 +21,31 @@ describe('WorkflowPublicationOutboxConsumer', () => { const reporter = mock(); const tracing = mock(); const eventService = mock(); + const abortSignal = new AbortController().signal; let consumer: WorkflowPublicationOutboxConsumer; const POLL_INTERVAL_MS = 15_000; + const LEASE_SECONDS = 120; + // Abort fires at 70% of the lease; abandonment follows after the grace period. + const ABORT_AFTER_MS = LEASE_SECONDS * 0.7 * 1000; + const ABANDON_GRACE_MS = 10_000; - function createConsumer(useWorkflowPublicationService = true, isLeader = true, concurrency = 1) { + let lifecycleLock: WorkflowPublicationLifecycleLock; + + function createConsumer( + useWorkflowPublicationService = true, + isLeader = true, + concurrency = 1, + leaseSeconds = LEASE_SECONDS, + ) { const workflowsConfig = mock({ useWorkflowPublicationService, publicationOutboxPollIntervalMs: POLL_INTERVAL_MS, workflowPublicationConcurrency: concurrency, + publicationOutboxLeaseSeconds: leaseSeconds, }); + lifecycleLock = new WorkflowPublicationLifecycleLock(); return new WorkflowPublicationOutboxConsumer( logger, workflowsConfig, @@ -40,7 +54,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { applier, reporter, mock({ isLeader }), - new WorkflowPublicationLifecycleLock(), + lifecycleLock, tracing, eventService, ); @@ -291,16 +305,19 @@ describe('WorkflowPublicationOutboxConsumer', () => { const result: PublicationResult = { type: 'completed', triggerStatuses: [] }; applier.apply.mockResolvedValue(result); - await consumer.processRecord(record); + await consumer.processRecord(record, abortSignal); - expect(applier.apply).toHaveBeenCalledWith(record); + expect(applier.apply).toHaveBeenCalledWith( + record, + expect.objectContaining({ signal: abortSignal }), + ); expect(reporter.report).toHaveBeenCalledWith(record, result); }); test('reports a failed result when the applier throws unexpectedly', async () => { applier.apply.mockRejectedValue(new Error('teardown failed')); - await consumer.processRecord(makeRecord()); + await consumer.processRecord(makeRecord(), abortSignal); expect(reporter.report).toHaveBeenCalledWith( expect.anything(), @@ -315,7 +332,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { const reportError = new Error('db write failed'); reporter.report.mockRejectedValue(reportError); - await expect(consumer.processRecord(makeRecord())).resolves.toBeUndefined(); + await expect(consumer.processRecord(makeRecord(), abortSignal)).resolves.toBeUndefined(); expect(errorReporter.error).toHaveBeenCalledWith(reportError, { shouldBeLogged: true }); }); @@ -324,7 +341,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { consumer = createConsumer(true, false); const record = makeRecord({ id: 7, workflowId: 'wf-7' }); - await consumer.processRecord(record); + await consumer.processRecord(record, abortSignal); expect(outboxRepository.returnToPending).toHaveBeenCalledWith(7); expect(applier.apply).not.toHaveBeenCalled(); @@ -332,6 +349,149 @@ describe('WorkflowPublicationOutboxConsumer', () => { }); }); + describe('abort and abandon', () => { + test('abandons a record that ignores the abort signal and keeps the drain alive', async () => { + const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null); + applier.apply.mockImplementationOnce(async () => await new Promise(() => {})); + consumer.startPolling(); + + const drain = consumer.drainPending(); + await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS + ABANDON_GRACE_MS); + + // The drain settles without the stuck record; no terminal status is + // written for it (it stays in_progress for lease reclaim). + await expect(drain).resolves.toBe(0); + expect(reporter.report).not.toHaveBeenCalled(); + expect(errorReporter.error).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('Abandoned workflow publication outbox record'), + }), + { shouldBeLogged: true }, + ); + + // A fresh drain still processes new records. + const next = makeRecord({ id: 2, workflowId: 'wf-2' }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(next).mockResolvedValue(null); + await expect(consumer.drainPending()).resolves.toBe(1); + expect(reporter.report).toHaveBeenCalledTimes(1); + }); + + test('a record that settles during the grace period still gets its terminal status', async () => { + const record = makeRecord({ id: 1 }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null); + // Settles after the abort deadline but within the grace period. + applier.apply.mockImplementationOnce( + async () => + await new Promise((resolve) => + setTimeout( + () => resolve({ type: 'completed', triggerStatuses: [] }), + ABORT_AFTER_MS + 1000, + ), + ), + ); + consumer.startPolling(); + + const drain = consumer.drainPending(); + await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS + 2000); + + await expect(drain).resolves.toBe(1); + expect(reporter.report).toHaveBeenCalledWith( + record, + expect.objectContaining({ type: 'completed' }), + ); + }); + + test('an abort honored by the applier still writes a terminal failed status', async () => { + const record = makeRecord({ id: 1 }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null); + applier.apply.mockImplementationOnce( + async (_record, abort) => + await new Promise((_resolve, reject) => { + abort?.signal.addEventListener('abort', () => reject(abort.signal.reason)); + }), + ); + consumer.startPolling(); + + const drain = consumer.drainPending(); + await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS); + + await expect(drain).resolves.toBe(1); + expect(reporter.report).toHaveBeenCalledWith( + record, + expect.objectContaining({ + type: 'failed', + error: expect.objectContaining({ + message: expect.stringContaining('exceeded its deadline'), + }), + }), + ); + }); + + test('holds the workflow lock until abandoned trigger operations settle', async () => { + const record = makeRecord({ id: 1, workflowId: 'wf-1' }); + let releaseDetached!: () => void; + const detachedWork = new Promise((resolve) => { + releaseDetached = resolve; + }); + applier.apply.mockImplementationOnce(async (_record, abort) => { + abort?.onDetached(detachedWork); + return { type: 'failed', error: new Error('deadline') }; + }); + const controller = new AbortController(); + + const processing = consumer.processRecord(record, controller.signal); + await vi.advanceTimersByTimeAsync(0); + + // The terminal status is written, but the lock is still held for the orphan. + expect(reporter.report).toHaveBeenCalledWith( + record, + expect.objectContaining({ type: 'failed' }), + ); + expect(lifecycleLock.isLocked('wf-1')).toBe(true); + + releaseDetached(); + await processing; + expect(lifecycleLock.isLocked('wf-1')).toBe(false); + }); + + test('leaves an aborted record in progress for lease reclaim instead of applying it', async () => { + const record = makeRecord({ id: 7, workflowId: 'wf-7' }); + const controller = new AbortController(); + controller.abort(); + + await consumer.processRecord(record, controller.signal); + + // Not returned to pending: the wait for the lock may have outlived the + // lease, and flipping the row would release a newer claimant's claim. + expect(outboxRepository.returnToPending).not.toHaveBeenCalled(); + expect(applier.apply).not.toHaveBeenCalled(); + expect(reporter.report).not.toHaveBeenCalled(); + }); + + test('scales the abandon grace down for short leases so abandonment stays within the lease', async () => { + // Lease 16s: abort fires at 11.2s, grace is capped at 4s (a quarter of + // the lease) instead of the fixed 10s, so abandonment lands at 15.2s — + // still inside the lease. + consumer = createConsumer(true, true, 1, 16); + const stuck = makeRecord({ id: 1 }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null); + applier.apply.mockImplementationOnce(async () => await new Promise(() => {})); + consumer.startPolling(); + + const drain = consumer.drainPending(); + await vi.advanceTimersByTimeAsync(15_200); + + await expect(drain).resolves.toBe(0); + expect(errorReporter.error).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining('Abandoned workflow publication outbox record'), + }), + { shouldBeLogged: true }, + ); + }); + }); + describe('metrics events', () => { function lastOutcome() { const calls = eventService.emit.mock.calls.filter( @@ -354,7 +514,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { async (result, expectedResult, expectedReason) => { applier.apply.mockResolvedValue(result); - await consumer.processRecord(makeRecord()); + await consumer.processRecord(makeRecord(), abortSignal); expect(lastOutcome()).toEqual( expect.objectContaining({ result: expectedResult, reason: expectedReason }), @@ -366,7 +526,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { applier.apply.mockResolvedValue({ type: 'completed', triggerStatuses: [] }); reporter.report.mockRejectedValue(new Error('db write failed')); - await consumer.processRecord(makeRecord()); + await consumer.processRecord(makeRecord(), abortSignal); expect(lastOutcome()).toEqual(expect.objectContaining({ result: 'failed', reason: 'none' })); }); @@ -374,7 +534,7 @@ describe('WorkflowPublicationOutboxConsumer', () => { test('does not emit when the record is returned to the queue (no longer leader)', async () => { consumer = createConsumer(true, false); - await consumer.processRecord(makeRecord()); + await consumer.processRecord(makeRecord(), abortSignal); expect(eventService.emit).not.toHaveBeenCalledWith( 'workflow-publication-outbox-record-processed', diff --git a/packages/cli/src/workflows/publication/workflow-publication-applier.ts b/packages/cli/src/workflows/publication/workflow-publication-applier.ts index 32f475543c1..1c546929b1e 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-applier.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-applier.ts @@ -22,6 +22,7 @@ import { WorkflowTriggerActivator, type TriggerActivationFailure, type TriggerActivationOutcome, + type TriggerOperationAbort, } from '@/workflows/triggers/workflow-trigger-activator'; import { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service'; @@ -81,7 +82,10 @@ export class WorkflowPublicationApplier { * must correspond to that version; otherwise the wrong triggers are * (de)registered. */ - async apply(record: WorkflowPublicationOutbox): Promise { + async apply( + record: WorkflowPublicationOutbox, + abort: TriggerOperationAbort, + ): Promise { const { workflow, oldVersion, newVersion } = await this.resolveVersions(record); if (!workflow) return { type: 'skipped', reason: 'workflow-not-found' }; @@ -90,12 +94,12 @@ export class WorkflowPublicationApplier { // A null `activeVersionId` means the workflow has been unpublished, so we // reconcile its triggers down to nothing rather than to a target version. if (workflow.activeVersionId === null) { - return await this.unpublish(workflow, oldVersion, record); + return await this.unpublish(workflow, oldVersion, record, abort); } if (!newVersion) return { type: 'version-missing' }; - return await this.publish(workflow, oldVersion, newVersion, record); + return await this.publish(workflow, oldVersion, newVersion, record, abort); } /** @@ -111,6 +115,7 @@ export class WorkflowPublicationApplier { oldVersion: WorkflowHistory | null, newVersion: WorkflowHistory, record: WorkflowPublicationOutbox, + abort: TriggerOperationAbort, ): Promise { const oldTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(oldVersion); const desiredTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(newVersion); @@ -155,16 +160,22 @@ export class WorkflowPublicationApplier { }; } + // Abort only before the trigger (de)activation phases — those run node + // code and can be slow; every state they leave behind on a mid-apply stop + // is one a crashed leader could also leave, which retries already handle. + abort.signal.throwIfAborted(); + // Must happen BEFORE advancing the version, using the currently published // version so the right webhooks are deregistered. A teardown failure here // bubbles up so the version is not advanced. if (toRemove.size > 0 && oldVersion) { - await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove); + await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort); } await this.advancePublishedVersion(record); try { + abort.signal.throwIfAborted(); if (toAdd.size > 0) { const activationMode = ACTIVATION_MODE_BY_REASON[record.reason ?? WorkflowPublicationReason.Publish]; @@ -173,6 +184,7 @@ export class WorkflowPublicationApplier { newVersion, toAdd, activationMode, + abort, ); return this.classifyActivationOutcome(outcome, desiredTriggerNodes, triggerKinds); } @@ -212,6 +224,7 @@ export class WorkflowPublicationApplier { workflow: WorkflowEntity, oldVersion: WorkflowHistory | null, record: WorkflowPublicationOutbox, + abort: TriggerOperationAbort, ): Promise { // If there is no oldVersion we may be retrying an unpublish that was // interrupted after removing the mapping: nothing to tear down, but we @@ -221,7 +234,8 @@ export class WorkflowPublicationApplier { ); if (oldVersion && toRemove.size > 0) { - await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove); + abort.signal.throwIfAborted(); + await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort); } // Invalidate before the mapping is removed, so reads fall through to the diff --git a/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts b/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts index d2d2d860b84..eb2624942c5 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts @@ -1,11 +1,12 @@ import { Logger } from '@n8n/backend-common'; import { WorkflowsConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; import { WorkflowPublicationOutbox, WorkflowPublicationOutboxRepository } from '@n8n/db'; import { OnLeaderTakeover, OnPubSubEvent, OnShutdown } from '@n8n/decorators'; import { Service } from '@n8n/di'; import { ErrorReporter, InstanceSettings, SpanStatus, Tracing } from 'n8n-core'; import { ensureError } from '@n8n/utils/errors/ensure-error'; -import { UnexpectedError } from 'n8n-workflow'; +import { OperationalError, UnexpectedError } from 'n8n-workflow'; import { EventService } from '@/events/event.service'; import type { @@ -16,6 +17,28 @@ import type { PublicationResult } from '@/workflows/publication/publication-resu import { PublicationStatusReporter } from '@/workflows/publication/publication-status-reporter'; import { WorkflowPublicationLifecycleLock } from '@/workflows/publication/workflow-publication-lifecycle-lock'; import { WorkflowPublicationApplier } from '@/workflows/publication/workflow-publication-applier'; +import type { TriggerOperationAbort } from '@/workflows/triggers/workflow-trigger-activator'; + +/** + * Resolved by {@link WorkflowPublicationOutboxConsumer.raceTimeout} when the + * timeout elapses before the raced promise settles. + */ +const TIMED_OUT = Symbol('timed-out'); + +/** + * Fraction of the outbox lease after which a record's processing is aborted. + * Below 1 so that an abandoned record is not reclaimable the moment its worker + * moves on — the same drain could otherwise claim it straight back. + */ +const ABORT_AFTER_LEASE_FRACTION = 0.7; + +/** + * How long an aborted record may take to unwind before it is abandoned, capped + * at a quarter of the lease: deadline plus grace must stay within the lease so + * a record is never still being waited on after it became reclaimable. + */ +const ABANDON_GRACE_MS = 10 * Time.seconds.toMilliseconds; +const ABANDON_GRACE_LEASE_FRACTION = 0.25; /** * Consumes the workflow publication outbox on the leader instance. It owns the @@ -165,20 +188,19 @@ export class WorkflowPublicationOutboxConsumer { let processed = 0; // Run worker loops in parallel. Each claims and processes records // until none remain (claiming is atomic, so workers never grab the same record). - let aborted = false; + let workerFailed = false; const runWorker = async () => { - while (!aborted && this.shouldKeepPolling()) { + while (!workerFailed && this.shouldKeepPolling()) { const record = await this.outboxRepository.claimNextPendingRecord(); if (!record) break; - await this.processRecord(record); - processed++; + if (await this.processRecordWithAbort(record)) processed++; } }; const workerTasks = Array.from({ length: concurrency }, async () => { await runWorker().catch((error) => { - aborted = true; + workerFailed = true; throw error; }); }); @@ -199,6 +221,59 @@ export class WorkflowPublicationOutboxConsumer { return this.isPolling && !this.isShuttingDown; } + /** + * Processes a record under a deadline: on expiry the record's abort signal + * fires, and after a further grace period the record is abandoned (left + * `in_progress` for lease reclaim) so a hung record never wedges the drain. + * Returns whether the record settled in time. + */ + private async processRecordWithAbort(record: WorkflowPublicationOutbox): Promise { + const leaseMs = + this.workflowsConfig.publicationOutboxLeaseSeconds * Time.seconds.toMilliseconds; + const abortAfterMs = leaseMs * ABORT_AFTER_LEASE_FRACTION; + const abandonGraceMs = Math.min(ABANDON_GRACE_MS, leaseMs * ABANDON_GRACE_LEASE_FRACTION); + + const controller = new AbortController(); + const work = this.processRecord(record, controller.signal); + + if ((await this.raceTimeout(work, abortAfterMs)) !== TIMED_OUT) return true; + + controller.abort(new OperationalError('Workflow publication processing exceeded its deadline')); + + if ((await this.raceTimeout(work, abandonGraceMs)) !== TIMED_OUT) return true; + + // A late rejection of the abandoned work must not become an unhandled rejection. + void work.catch((error) => + this.errorReporter.error(ensureError(error), { shouldBeLogged: true }), + ); + this.errorReporter.error( + new OperationalError( + 'Abandoned workflow publication outbox record: processing did not stop after abort', + { extra: { outboxId: record.id, workflowId: record.workflowId } }, + ), + { shouldBeLogged: true }, + ); + return false; + } + + /** Resolves with the raced promise, or with {@link TIMED_OUT} after `timeoutMs`. */ + private async raceTimeout( + promise: Promise, + timeoutMs: number, + ): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((resolve) => { + timer = setTimeout(() => resolve(TIMED_OUT), timeoutMs); + }), + ]); + } finally { + clearTimeout(timer); + } + } + /** * Applies a single claimed record and reports the outcome. The applier returns * a `failed` result for expected failures; an unexpected throw is wrapped into @@ -212,7 +287,7 @@ export class WorkflowPublicationOutboxConsumer { * between claiming the record and entering the critical section, the record is returned * to the queue (so the new leader reprocesses it) and nothing is applied here. */ - async processRecord(record: WorkflowPublicationOutbox): Promise { + async processRecord(record: WorkflowPublicationOutbox, signal: AbortSignal): Promise { await this.tracing.startSpan( { name: 'Publication outbox record', @@ -237,6 +312,19 @@ export class WorkflowPublicationOutboxConsumer { return; } + // The worker may have aborted this record while it queued on the lock; + // starting now would apply work long after it was abandoned. The row is + // left `in_progress` for lease reclaim rather than returned to pending: + // the wait may have outlived the lease, and flipping the row here would + // release the claim of whichever worker has since reclaimed it. + if (signal.aborted) { + this.logger.debug('Skipped applying publication outbox record: aborted', { + outboxId: record.id, + workflowId: record.workflowId, + }); + return; + } + this.logger.debug('Started processing workflow publication outbox record', { outboxId: record.id, workflowId: record.workflowId, @@ -246,13 +334,25 @@ export class WorkflowPublicationOutboxConsumer { const startedAt = Date.now(); let result: PublicationResult; + // An aborted per-node operation is abandoned, not cancelled; collect + // every orphan so the lock outlives whatever may still mutate this + // workflow's registrations. + const detachedWork: Array> = []; + const abort: TriggerOperationAbort = { + signal, + onDetached: (work) => detachedWork.push(work), + }; + try { - result = await this.applier.apply(record); + result = await this.applier.apply(record, abort); } catch (error) { const cause = ensureError(error); result = { type: 'failed', - error: new UnexpectedError(`Unexpected: ${cause.message}`, { cause }), + // An abort is our own doing, not an unexpected applier failure. + error: signal.aborted + ? cause + : new UnexpectedError(`Unexpected: ${cause.message}`, { cause }), }; } @@ -276,6 +376,17 @@ export class WorkflowPublicationOutboxConsumer { }); span.setAttribute('n8n.publication.result', result.type); + + // The terminal status is already written; keep the lock until the + // abandoned operations settle so the next record for this workflow + // can never run concurrently with them. + if (detachedWork.length > 0) { + this.logger.warn( + 'Keeping workflow publication lock held until abandoned trigger operations settle', + { outboxId: record.id, workflowId: record.workflowId }, + ); + await Promise.allSettled(detachedWork); + } }); span.setStatus({ code: SpanStatus.ok }); diff --git a/packages/cli/src/workflows/triggers/__tests__/trigger-activation-retry.test.ts b/packages/cli/src/workflows/triggers/__tests__/trigger-activation-retry.test.ts index 2590fc14e59..32b8f520490 100644 --- a/packages/cli/src/workflows/triggers/__tests__/trigger-activation-retry.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/trigger-activation-retry.test.ts @@ -1,3 +1,5 @@ +import { ensureError } from '@n8n/utils/errors/ensure-error'; +import { sleep } from '@n8n/utils/sleep'; import { WebhookPathTakenError } from 'n8n-workflow'; import { @@ -9,15 +11,19 @@ vi.mock('@n8n/utils/sleep', () => ({ sleep: vi.fn(), })); +const flushPromises = async () => await new Promise((resolve) => setImmediate(resolve)); + const MAX_ATTEMPTS = 3; describe('retryTriggerActivation', () => { + const signal = new AbortController().signal; + beforeEach(() => vi.clearAllMocks()); test('resolves without retrying when activation succeeds', async () => { const activate = vi.fn().mockResolvedValue(undefined); - await retryTriggerActivation(activate, MAX_ATTEMPTS); + await retryTriggerActivation(activate, MAX_ATTEMPTS, signal); expect(activate).toHaveBeenCalledTimes(1); }); @@ -28,7 +34,7 @@ describe('retryTriggerActivation', () => { .mockRejectedValueOnce(new Error('transient')) .mockResolvedValueOnce(undefined); - await retryTriggerActivation(activate, MAX_ATTEMPTS); + await retryTriggerActivation(activate, MAX_ATTEMPTS, signal); expect(activate).toHaveBeenCalledTimes(2); }); @@ -37,7 +43,7 @@ describe('retryTriggerActivation', () => { const error = new Error('transient'); const activate = vi.fn().mockRejectedValue(error); - await expect(retryTriggerActivation(activate, MAX_ATTEMPTS)).rejects.toBe(error); + await expect(retryTriggerActivation(activate, MAX_ATTEMPTS, signal)).rejects.toBe(error); expect(activate).toHaveBeenCalledTimes(MAX_ATTEMPTS); }); @@ -45,9 +51,30 @@ describe('retryTriggerActivation', () => { const error = new WebhookPathTakenError('Webhook'); const activate = vi.fn().mockRejectedValue(error); - await expect(retryTriggerActivation(activate, MAX_ATTEMPTS)).rejects.toBe(error); + await expect(retryTriggerActivation(activate, MAX_ATTEMPTS, signal)).rejects.toBe(error); expect(activate).toHaveBeenCalledTimes(1); }); + + test('stops sleeping and rethrows the abort reason when the signal fires during backoff', async () => { + vi.mocked(sleep).mockImplementationOnce( + async (_ms, sleepSignal) => + await new Promise((_resolve, reject) => { + sleepSignal?.addEventListener('abort', () => reject(ensureError(sleepSignal.reason)), { + once: true, + }); + }), + ); + const activate = vi.fn().mockRejectedValue(new Error('transient')); + const controller = new AbortController(); + + const retrying = retryTriggerActivation(activate, MAX_ATTEMPTS, controller.signal); + await flushPromises(); + controller.abort(new Error('deadline')); + + await expect(retrying).rejects.toThrow('deadline'); + expect(activate).toHaveBeenCalledTimes(1); + expect(sleep).toHaveBeenCalledWith(expect.any(Number), controller.signal); + }); }); describe('isTransientActivationError', () => { diff --git a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts index b6e2648fdae..832541bd69e 100644 --- a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts @@ -31,6 +31,8 @@ const MAX_ATTEMPTS = TRIGGER_ACTIVATION_MAX_ATTEMPTS; const flushPromises = async () => await new Promise((resolve) => setImmediate(resolve)); +const abort = { signal: new AbortController().signal, onDetached: vi.fn() }; + const tracing = mock(); const eventService = mock(); @@ -286,6 +288,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['t', 'p', 'webhook-node']), 'update', + abort, ); // Both phases overlap inside one isolate bracket, so their relative order is @@ -336,6 +339,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['t', 'webhook-node']), 'init', + abort, ); expect(nonWebhookTriggerRegistrar.createRegistrationContext).toHaveBeenCalledWith( @@ -391,6 +395,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['webhook-node', 'trigger-node']), 'update', + abort, ), ).rejects.toThrow('webhook discovery failed'); @@ -438,6 +443,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-node', 'trigger-a', 'trigger-b']), + abort, ) .then(() => { deactivateSettled = true; @@ -505,6 +511,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-node', 'trigger-node']), + abort, ), ).rejects.toThrow('webhook discovery failed'); @@ -548,6 +555,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['webhook-ok', 'webhook-bad', 't', 'p']), 'update', + abort, ); // Both registrars ran; each phase surfaced its own failure while keeping the @@ -592,6 +600,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['webhook-a', 'webhook-b']), 'update', + abort, ); // Parallel fan-out: assert by membership, not order. @@ -632,6 +641,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('webhook-a', 'webhook', { name: 'Webhook A' })], connections: {} }, new Set(['webhook-a']), 'update', + abort, ); expect(outcome).toEqual({ activated: ['webhook-a'], failures: [] }); @@ -678,6 +688,7 @@ describe('WorkflowTriggerActivator', () => { }, new Set(['webhook-a', 'webhook-b']), 'update', + abort, ); // Parallel fan-out: assert by membership, not order. @@ -719,6 +730,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} }, new Set(['webhook-node']), 'update', + abort, ); expect(outcome.activated).toEqual([]); @@ -757,6 +769,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('t', 'trigger'), node('p', 'poll')], connections: {} }, new Set(['t', 'p']), 'update', + abort, ); expect(outcome.activated).toEqual(expect.arrayContaining(['t'])); @@ -799,6 +812,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('p', 'poll')], connections: {} }, new Set(['p']), 'update', + abort, ); expect(outcome).toEqual({ activated: ['p'], failures: [] }); @@ -839,6 +853,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('t', 'trigger')], connections: {} }, new Set(['t']), 'update', + abort, ); const context = deps.nonWebhookTriggerRegistrar.createRegistrationContext.mock.calls[0][1]; @@ -982,6 +997,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('webhook-a', 'webhook', { name: 'Webhook A' })], connections: {} }, new Set(['webhook-a']), 'update', + abort, ); expect(getEmissions('workflow-publication-trigger-operation')).toContainEqual( @@ -1011,6 +1027,7 @@ describe('WorkflowTriggerActivator', () => { { nodes: [node('webhook-b', 'webhook', { name: 'Webhook B' })], connections: {} }, new Set(['webhook-b']), 'update', + abort, ); expect(getEmissions('workflow-publication-trigger-operation')).toContainEqual( @@ -1036,6 +1053,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('trigger-a', 'trigger')], connections: {} }, new Set(['trigger-a']), + abort, ); expect(getEmissions('workflow-publication-trigger-operation')).toContainEqual( @@ -1046,4 +1064,122 @@ describe('WorkflowTriggerActivator', () => { ); }); }); + + describe('abort', () => { + beforeEach(() => { + vi.spyOn(WorkflowExpression.prototype, 'acquireIsolate').mockResolvedValue(true); + vi.spyOn(WorkflowExpression.prototype, 'releaseIsolate').mockResolvedValue(); + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue( + mock(), + ); + }); + + test('records a node whose registration hangs as failed once the signal aborts, keeping other nodes', async () => { + const webhookTriggerRegistrar = mock(); + webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.createRegistrationContext.mockReturnValue( + mock(), + ); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['ok', 'stuck']); + nonWebhookTriggerRegistrar.register.mockImplementation(async (_workflow, _reg, nodeId) => { + if (nodeId === 'stuck') await new Promise(() => {}); + }); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const controller = new AbortController(); + const onDetached = vi.fn(); + + const activation = activator.activate( + mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), + { nodes: [node('ok', 'trigger'), node('stuck', 'trigger')], connections: {} }, + new Set(['ok', 'stuck']), + 'update', + { signal: controller.signal, onDetached }, + ); + await flushPromises(); + controller.abort(new Error('deadline')); + + const outcome = await activation; + expect(outcome.activated).toEqual(['ok']); + expect(outcome.failures).toEqual([ + expect.objectContaining({ + nodeId: 'stuck', + error: expect.objectContaining({ message: 'deadline' }), + }), + ]); + // The hung registration is handed back so the caller can outlive it. + expect(onDetached).toHaveBeenCalledTimes(1); + expect(onDetached).toHaveBeenCalledWith(expect.any(Promise)); + }); + + test('a deactivation whose teardown hangs rejects with the abort reason once the signal fires', async () => { + const webhookTriggerRegistrar = mock(); + webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['stuck']); + nonWebhookTriggerRegistrar.deregister.mockImplementation( + async () => await new Promise(() => {}), + ); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const controller = new AbortController(); + const onDetached = vi.fn(); + + const deactivation = activator.deactivate( + mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), + { nodes: [node('stuck', 'trigger')], connections: {} }, + new Set(['stuck']), + { signal: controller.signal, onDetached }, + ); + await flushPromises(); + controller.abort(new Error('deadline')); + + await expect(deactivation).rejects.toThrow('deadline'); + expect(onDetached).toHaveBeenCalledWith(expect.any(Promise)); + }); + + test('clears webhook rows only for nodes whose every webhook deregistered before the abort', async () => { + const webhookTriggerRegistrar = mock(); + const webhookOk = mock({ node: 'Webhook OK', path: 'ok' }); + // The stuck node has two webhooks; only one of them hangs. Its fulfilled + // sibling must not cause the node's rows to be cleared. + const webhookStuck = mock({ node: 'Webhook Stuck', path: 'hang' }); + const webhookStuckSibling = mock({ node: 'Webhook Stuck', path: 'fine' }); + webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([ + webhookOk, + webhookStuck, + webhookStuckSibling, + ]); + webhookTriggerRegistrar.deregister.mockImplementation(async ({ webhookData }) => { + if (webhookData.path === 'hang') await new Promise(() => {}); + return webhookData.node; + }); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const controller = new AbortController(); + + const deactivation = activator.deactivate( + mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), + { + nodes: [ + node('ok-node', 'webhook', { name: 'Webhook OK' }), + node('stuck-node', 'webhook', { name: 'Webhook Stuck' }), + ], + connections: {}, + }, + new Set(['ok-node', 'stuck-node']), + { signal: controller.signal, onDetached: vi.fn() }, + ); + await flushPromises(); + controller.abort(new Error('deadline')); + + await expect(deactivation).rejects.toThrow('deadline'); + expect(webhookTriggerRegistrar.clearWorkflowWebhooksForNodes).toHaveBeenCalledWith('wf-1', [ + 'Webhook OK', + ]); + }); + }); }); diff --git a/packages/cli/src/workflows/triggers/trigger-activation-retry.ts b/packages/cli/src/workflows/triggers/trigger-activation-retry.ts index 32b1b70d172..622c7f04925 100644 --- a/packages/cli/src/workflows/triggers/trigger-activation-retry.ts +++ b/packages/cli/src/workflows/triggers/trigger-activation-retry.ts @@ -12,7 +12,9 @@ export const isTransientActivationError = (error: Error): boolean => /** * Activates a single trigger node, retrying transient failures in-process with - * exponential backoff up to `maxAttempts` before giving up. + * exponential backoff up to `maxAttempts` before giving up. Once `signal` + * aborts, no further attempt is started: a caller that abandoned this + * activation must not have an old registration committed behind its back. * * The activate function must be self-atomic — it must leave no partial state behind on * failure — so a re-attempt does not conflict with itself and needs no cleanup. @@ -20,20 +22,27 @@ export const isTransientActivationError = (error: Error): boolean => export async function retryTriggerActivation( activate: () => Promise, maxAttempts: number, + signal: AbortSignal, ): Promise { for (let attempt = 0; ; attempt++) { + signal.throwIfAborted(); try { await activate(); return; } catch (error) { const isLastAttempt = attempt >= maxAttempts - 1; - if (!isTransientActivationError(ensureError(error)) || isLastAttempt) throw error; + if (!isTransientActivationError(ensureError(error)) || isLastAttempt || signal.aborted) + throw error; + // `sleep` rejects with the abort reason as soon as the signal fires, so + // an abandoned retry unwinds promptly instead of sleeping out the backoff + // (up to a day) while its caller holds the workflow's lifecycle lock. await sleep( Math.min( WORKFLOW_REACTIVATE_INITIAL_TIMEOUT * 2 ** attempt, WORKFLOW_REACTIVATE_MAX_TIMEOUT, ), + signal, ); } } diff --git a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts index 69827ae0ce7..133a71d4bb8 100644 --- a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts +++ b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts @@ -46,6 +46,45 @@ import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.serv export type { WorkflowTriggerVersion }; +/** + * How callers request that trigger (de)activation be abortable. Node + * registration and teardown code cannot be cancelled, only abandoned, and an + * abandoned operation may still mutate this workflow's registrations when it + * eventually settles. `onDetached` hands every such orphan back to the caller, + * which must not release the workflow's lifecycle lock until they settle. + */ +export interface TriggerOperationAbort { + signal: AbortSignal; + onDetached: (work: Promise) => void; +} + +/** + * Resolves with `promise`, or rejects with the abort reason once the signal + * fires — reporting the abandoned promise through `onDetached`. + */ +async function raceAbort(promise: Promise, abort: TriggerOperationAbort): Promise { + const { signal } = abort; + if (signal.aborted) { + abort.onDetached(promise); + throw ensureError(signal.reason); + } + + let onAbort!: () => void; + const aborted = new Promise((_resolve, reject) => { + onAbort = () => reject(ensureError(signal.reason)); + signal.addEventListener('abort', onAbort, { once: true }); + }); + try { + return await Promise.race([promise, aborted]); + } catch (error) { + // Reporting a promise that lost to its own rejection is harmless. + if (signal.aborted) abort.onDetached(promise); + throw error; + } finally { + signal.removeEventListener('abort', onAbort); + } +} + // Their trigger() is a no-op — fired by the execution engine, never the // registry — so reconciling them against the registry would re-enqueue forever. const PSEUDO_TRIGGER_NODE_TYPES = new Set([ @@ -232,10 +271,17 @@ export class WorkflowTriggerActivator { version: WorkflowTriggerVersion, nodeIds: Set, activationMode: WorkflowActivateMode, + abort: TriggerOperationAbort, ): Promise { const startedAt = Date.now(); try { - const outcome = await this.activateInternal(dbWorkflow, version, nodeIds, activationMode); + const outcome = await this.activateInternal( + dbWorkflow, + version, + nodeIds, + activationMode, + abort, + ); this.emitTriggerOperation( 'activate', outcome.failures.length === 0 ? 'success' : 'failure', @@ -254,6 +300,7 @@ export class WorkflowTriggerActivator { version: WorkflowTriggerVersion, nodeIds: Set, activationMode: WorkflowActivateMode, + abort: TriggerOperationAbort, ): Promise { return await this.tracing.startSpan( { @@ -289,6 +336,7 @@ export class WorkflowTriggerActivator { nodeIds, outcome, activationMode, + abort, ), this.registerNonWebhookTriggers( dbWorkflow, @@ -298,6 +346,7 @@ export class WorkflowTriggerActivator { nodeIds, outcome, activationMode, + abort, ), ]); this.throwRejectedPhaseError(phaseResults); @@ -333,12 +382,13 @@ export class WorkflowTriggerActivator { dbWorkflow: WorkflowEntity, version: WorkflowTriggerVersion, nodeIds: Set, + abort: TriggerOperationAbort, ) { if (nodeIds.size === 0) return; const startedAt = Date.now(); try { - await this.deactivateInternal(dbWorkflow, version, nodeIds); + await this.deactivateInternal(dbWorkflow, version, nodeIds, abort); this.emitTriggerOperation('deactivate', 'success', startedAt); this.emitTriggerNodeOperations('deactivate', nodeIds.size, 0); } catch (error) { @@ -352,6 +402,7 @@ export class WorkflowTriggerActivator { dbWorkflow: WorkflowEntity, version: WorkflowTriggerVersion, nodeIds: Set, + abort: TriggerOperationAbort, ) { await this.tracing.startSpan( { @@ -374,14 +425,8 @@ export class WorkflowTriggerActivator { // The non-webhook phase doesn't touch the expression isolate that the // webhook deregister acquires, so the two phases can overlap. const phaseResults = await Promise.allSettled([ - this.deregisterWebhookTriggers(workflow, additionalData, nodeIds).then( - async (removedNodeNames) => - await this.webhookTriggerRegistrar.clearWorkflowWebhooksForNodes( - dbWorkflow.id, - removedNodeNames, - ), - ), - this.deregisterNonWebhookTriggers(dbWorkflow.id, workflow, nodeIds), + this.deregisterWebhookTriggers(workflow, additionalData, nodeIds, abort), + this.deregisterNonWebhookTriggers(dbWorkflow.id, workflow, nodeIds, abort), ]); this.throwRejectedPhaseError(phaseResults); @@ -492,6 +537,7 @@ export class WorkflowTriggerActivator { nodeIds: Set, outcome: TriggerActivationOutcome, activationMode: WorkflowActivateMode, + abort: TriggerOperationAbort, ) { const webhooksByNode = this.groupWebhookTriggersByNode(workflow, additionalData, nodeIds); @@ -503,6 +549,7 @@ export class WorkflowTriggerActivator { nodeName, webhooks, activationMode, + abort, ), ); @@ -523,18 +570,23 @@ export class WorkflowTriggerActivator { nodeName: string, webhooks: IWebhookData[], activationMode: WorkflowActivateMode, + abort: TriggerOperationAbort, ): Promise> { try { for (const webhookData of webhooks) { - await retryTriggerActivation( - async () => - await this.webhookTriggerRegistrar.register({ - workflow, - webhookData, - mode: 'trigger', - activation: activationMode, - }), - TRIGGER_ACTIVATION_MAX_ATTEMPTS, + await raceAbort( + retryTriggerActivation( + async () => + await this.webhookTriggerRegistrar.register({ + workflow, + webhookData, + mode: 'trigger', + activation: activationMode, + }), + TRIGGER_ACTIVATION_MAX_ATTEMPTS, + abort.signal, + ), + abort, ); } return createResultOk({ nodeId }); @@ -575,8 +627,10 @@ export class WorkflowTriggerActivator { workflow: Workflow, additionalData: IWorkflowExecuteAdditionalData, nodeIds: Set, + abort: TriggerOperationAbort, ) { const removedNodeNames: string[] = []; + let firstFailure: Error | undefined; await workflow.expression.acquireIsolate(); try { @@ -585,18 +639,43 @@ export class WorkflowTriggerActivator { const deregistrationResults = await Promise.allSettled( webhooks.map( async (webhookData) => - await this.webhookTriggerRegistrar.deregister({ workflow, webhookData }), + await raceAbort( + this.webhookTriggerRegistrar.deregister({ workflow, webhookData }), + abort, + ), ), ); - for (const result of deregistrationResults) { - if (result.status === 'rejected') throw ensureError(result.reason); - removedNodeNames.push(result.value); + // Row cleanup is per node, so only a node whose EVERY webhook + // deregistered may be cleared: a failed or abandoned webhook's row is + // the only record left for deregistering it externally later. + const pendingWebhooksByNode = new Map(); + for (const webhookData of webhooks) { + pendingWebhooksByNode.set( + webhookData.node, + (pendingWebhooksByNode.get(webhookData.node) ?? 0) + 1, + ); } + deregistrationResults.forEach((result, index) => { + if (result.status === 'rejected') { + firstFailure ??= ensureError(result.reason); + return; + } + const nodeName = webhooks[index].node; + const pending = (pendingWebhooksByNode.get(nodeName) ?? 0) - 1; + pendingWebhooksByNode.set(nodeName, pending); + if (pending === 0) removedNodeNames.push(nodeName); + }); } finally { await workflow.expression.releaseIsolate(); } + // Clear rows for the nodes that fully deregistered even when another + // node's teardown failed or was abandoned, so their `webhook_entity` rows + // don't outlive the deregistration. + await this.webhookTriggerRegistrar.clearWorkflowWebhooksForNodes(workflow.id, removedNodeNames); + if (firstFailure) throw firstFailure; + await this.workflowStaticDataService.saveStaticData(workflow); return removedNodeNames; @@ -627,6 +706,7 @@ export class WorkflowTriggerActivator { nodeIds: Set, outcome: TriggerActivationOutcome, activationMode: WorkflowActivateMode, + abort: TriggerOperationAbort, ) { const triggerNodeIds = this.getNonWebhookTriggerNodeIdsForNodeIds(workflow, nodeIds); if (triggerNodeIds.length === 0) return; @@ -652,10 +732,16 @@ export class WorkflowTriggerActivator { const results = await Promise.all( triggerNodeIds.map(async (nodeId): Promise> => { try { - await retryTriggerActivation( - async () => - await this.nonWebhookTriggerRegistrar.register(workflow, registration, nodeId), - TRIGGER_ACTIVATION_MAX_ATTEMPTS, + // NOTE: to abort the actual trigger operation, we would need to pass the signal all the way + // down to the node. This doesn't happen today, but could in the future. + await raceAbort( + retryTriggerActivation( + async () => + await this.nonWebhookTriggerRegistrar.register(workflow, registration, nodeId), + TRIGGER_ACTIVATION_MAX_ATTEMPTS, + abort.signal, + ), + abort, ); return createResultOk(nodeId); @@ -690,12 +776,14 @@ export class WorkflowTriggerActivator { workflowId: WorkflowId, workflow: Workflow, nodeIds: Set, + abort: TriggerOperationAbort, ) { const triggerNodeIds = this.getNonWebhookTriggerNodeIdsForNodeIds(workflow, nodeIds); await Promise.all( triggerNodeIds.map( - async (nodeId) => await this.nonWebhookTriggerRegistrar.deregister(workflowId, nodeId), + async (nodeId) => + await raceAbort(this.nonWebhookTriggerRegistrar.deregister(workflowId, nodeId), abort), ), ); } @@ -759,14 +847,19 @@ export class WorkflowTriggerActivator { this.triggerExecutionContextFactory.executeErrorWorkflow(activationError, workflowData, mode); // `addTriggers` does not own the expression isolate, so acquire it per attempt. - await retryTriggerActivation(async () => { - await workflow.expression.acquireIsolate(); - try { - await this.nonWebhookTriggerRegistrar.register(workflow, registration, node.id); - } finally { - await workflow.expression.releaseIsolate(); - } - }, TRIGGER_ACTIVATION_MAX_ATTEMPTS); + await retryTriggerActivation( + async () => { + await workflow.expression.acquireIsolate(); + try { + await this.nonWebhookTriggerRegistrar.register(workflow, registration, node.id); + } finally { + await workflow.expression.releaseIsolate(); + } + }, + TRIGGER_ACTIVATION_MAX_ATTEMPTS, + // Runtime reactivation has no abort context; never aborts. + new AbortController().signal, + ); await this.workflowStaticDataService.saveStaticData(workflow); diff --git a/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts b/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts index 355fa4ee8d8..fe4c4baa167 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts @@ -38,6 +38,8 @@ mockInstance(WorkflowService); mockInstance(OwnershipService); mockInstance(ExternalHooks); +const abortSignal = new AbortController().signal; + let consumer: WorkflowPublicationOutboxConsumer; let activeWorkflowManager: ActiveWorkflowManager; let activeWorkflowTriggers: ActiveWorkflowTriggers; @@ -125,7 +127,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { const record = await outboxRepository.claimNextPendingRecord(); expect(record).not.toBeNull(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // Surgical in-memory result: unchanged kept, removed gone, added registered. const state = activeWorkflowTriggers.get(workflow.id); @@ -165,7 +167,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // `missing` got re-registered; `present` was left untouched (same response object). const state = activeWorkflowTriggers.get(workflow.id); @@ -192,7 +194,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // Nothing re-registered (same response object) and the version is unchanged. expect(activeWorkflowTriggers.get(workflow.id)?.get(trigger.id)).toBe(responseBefore); @@ -226,7 +228,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); const published = await publishedVersionRepository.getPublishedVersionWithRelations( diff --git a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts index 0cc373316ea..fbc2c4be782 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts @@ -43,6 +43,8 @@ mockInstance(OwnershipService); mockInstance(ExternalHooks); let reconciler: WorkflowPublicationReconciler; +const abortSignal = new AbortController().signal; + let consumer: WorkflowPublicationOutboxConsumer; let activeWorkflowTriggers: ActiveWorkflowTriggers; let outboxRepository: WorkflowPublicationOutboxRepository; @@ -123,7 +125,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // the reporter persists the `activated` trigger-status rows with kinds. await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); // The leader-transition race: a demoted main consumed the outbox record @@ -154,7 +156,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // A fresh boot: the registry is empty, the published state is intact. await activeWorkflowTriggers.remove(workflow.id); @@ -197,7 +199,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // `activated` trigger-status rows. await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // An unpublish interrupted after removing the published-version mapping // but before the reporter cleared the trigger-status rows, with its outbox @@ -227,7 +229,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); // A demoted main consumed the unpublish record: workflow deactivated, @@ -254,7 +256,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // A re-leased unpublish torn between two mains: mapping removed and the // record completed elsewhere, but this leader's registry AND the @@ -280,7 +282,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); // Mid-unpublish: activeVersionId already cleared, pending record owns the // teardown. Reconciliation must not race it. @@ -300,7 +302,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // A parameter-only newer version is published: the trigger node set is // identical, so no node-id diff can distinguish the two versions. @@ -311,7 +313,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { }); await setActiveVersion(workflow.id, newVersionId); await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBe(newVersionId); // A stalled processor (zombie writer) rolls the mapping back to the old @@ -336,7 +338,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // An unpublish fully applied elsewhere (triggers down, status rows // cleared, record terminal), after which a zombie writer restored the @@ -363,7 +365,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // Mid-flight publish of a parameter-only new version: `activeVersionId` // commits together with the pending record, and the mapping still points @@ -400,7 +402,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); const newVersionId = 'version-2-status-drift'; await createWorkflowHistory(workflow, owner, undefined, { @@ -409,7 +411,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { }); await setActiveVersion(workflow.id, newVersionId); await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // A zombie writer rewrites the status rows for the old version after its // record already resolved. The mapping still agrees with `activeVersionId`, @@ -478,7 +480,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // A v2 publish that crashed after advancing the mapping but before the // reporter rewrote the rows: mapping equals `activeVersionId`, the record @@ -511,7 +513,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const drifted = await createWorkflowWithHistory({ active: true, nodes: [driftTrigger] }, owner); await setActiveVersion(drifted.id, drifted.versionId); await outboxRepository.enqueue(drifted.id, drifted.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); const newVersionId = 'version-2-drift-in-flight'; await createWorkflowHistory(drifted, owner, undefined, { versionId: newVersionId, @@ -552,7 +554,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await setActiveVersion(workflow.id, workflow.versionId); await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); // Activation is untouched: a genuine publish still registers the no-op // trigger's registry slot, and the status row (whose presence drives the @@ -584,7 +586,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); - await consumer.processRecord(record!); + await consumer.processRecord(record!, abortSignal); const registeredBefore = activeWorkflowTriggers.get(workflow.id)?.get(trigger.id); expect(registeredBefore).toBeDefined();