fix(core): Bound workflow publication outbox record processing with an abort deadline (backport to release-candidate/2.35.x) (#36286)

Co-authored-by: mfsiega <93014743+mfsiega@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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<typeof callerPolicySchema>;
 
+// 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. */

packages/@n8n/config/test/config.test.ts

@@ -978,6 +978,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', () => {

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));
+	});
 });

packages/@n8n/utils/src/sleep.ts

@@ -1,25 +1,29 @@
+import { ensureError } from './errors/ensure-error';
+
 async function sleepWithAbort(ms: number, abortSignal: AbortSignal): Promise<void> {
 	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<void> {
 	if (!abortSignal) {

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();
+		});
+	});
 });

packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts

@@ -21,17 +21,31 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 	const reporter = mock<PublicationStatusReporter>();
 	const tracing = mock<Tracing>();
 	const eventService = mock<EventService>();
+	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<WorkflowsConfig>({
 			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<InstanceSettings>({ 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<void>((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',

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<PublicationResult> {
+	async apply(
+		record: WorkflowPublicationOutbox,
+		abort: TriggerOperationAbort,
+	): Promise<PublicationResult> {
 		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<PublicationResult> {
 		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<PublicationResult> {
 		// 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

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<boolean> {
+		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<T>(
+		promise: Promise<T>,
+		timeoutMs: number,
+	): Promise<T | typeof TIMED_OUT> {
+		let timer: NodeJS.Timeout | undefined;
+		try {
+			return await Promise.race([
+				promise,
+				new Promise<typeof TIMED_OUT>((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<void> {
+	async processRecord(record: WorkflowPublicationOutbox, signal: AbortSignal): Promise<void> {
 		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<Promise<unknown>> = [];
+					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 });

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', () => {

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<Tracing>();
 const eventService = mock<EventService>();
 
@@ -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<WorkflowEntity>({ 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<IWorkflowExecuteAdditionalData>(),
+			);
+		});
+
+		test('records a node whose registration hangs as failed once the signal aborts, keeping other nodes', async () => {
+			const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
+			webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]);
+			const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
+			nonWebhookTriggerRegistrar.createRegistrationContext.mockReturnValue(
+				mock<PreparedNonWebhookTriggerRegistration>(),
+			);
+			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<WorkflowEntity>({ 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>();
+			webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]);
+			const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
+			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<WorkflowEntity>({ 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<WebhookTriggerRegistrar>();
+			const webhookOk = mock<IWebhookData>({ 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<IWebhookData>({ node: 'Webhook Stuck', path: 'hang' });
+			const webhookStuckSibling = mock<IWebhookData>({ 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>();
+			nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
+
+			const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
+			const controller = new AbortController();
+
+			const deactivation = activator.deactivate(
+				mock<WorkflowEntity>({ 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',
+			]);
+		});
+	});
 });

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<void>,
 	maxAttempts: number,
+	signal: AbortSignal,
 ): Promise<void> {
 	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,
 			);
 		}
 	}

packages/cli/src/workflows/triggers/workflow-trigger-activator.ts

@@ -45,6 +45,45 @@ import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.serv
 
 export type WorkflowTriggerVersion = { nodes: INode[]; connections: IConnections };
 
+/**
+ * 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<unknown>) => void;
+}
+
+/**
+ * Resolves with `promise`, or rejects with the abort reason once the signal
+ * fires — reporting the abandoned promise through `onDetached`.
+ */
+async function raceAbort<T>(promise: Promise<T>, abort: TriggerOperationAbort): Promise<T> {
+	const { signal } = abort;
+	if (signal.aborted) {
+		abort.onDetached(promise);
+		throw ensureError(signal.reason);
+	}
+
+	let onAbort!: () => void;
+	const aborted = new Promise<never>((_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<string>([
@@ -245,10 +284,17 @@ export class WorkflowTriggerActivator {
 		version: WorkflowTriggerVersion,
 		nodeIds: Set<INode['id']>,
 		activationMode: WorkflowActivateMode,
+		abort: TriggerOperationAbort,
 	): Promise<TriggerActivationOutcome> {
 		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',
@@ -267,6 +313,7 @@ export class WorkflowTriggerActivator {
 		version: WorkflowTriggerVersion,
 		nodeIds: Set<INode['id']>,
 		activationMode: WorkflowActivateMode,
+		abort: TriggerOperationAbort,
 	): Promise<TriggerActivationOutcome> {
 		return await this.tracing.startSpan(
 			{
@@ -302,6 +349,7 @@ export class WorkflowTriggerActivator {
 							nodeIds,
 							outcome,
 							activationMode,
+							abort,
 						),
 						this.registerNonWebhookTriggers(
 							dbWorkflow,
@@ -311,6 +359,7 @@ export class WorkflowTriggerActivator {
 							nodeIds,
 							outcome,
 							activationMode,
+							abort,
 						),
 					]);
 					this.throwRejectedPhaseError(phaseResults);
@@ -346,12 +395,13 @@ export class WorkflowTriggerActivator {
 		dbWorkflow: WorkflowEntity,
 		version: WorkflowTriggerVersion,
 		nodeIds: Set<INode['id']>,
+		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) {
@@ -365,6 +415,7 @@ export class WorkflowTriggerActivator {
 		dbWorkflow: WorkflowEntity,
 		version: WorkflowTriggerVersion,
 		nodeIds: Set<INode['id']>,
+		abort: TriggerOperationAbort,
 	) {
 		await this.tracing.startSpan(
 			{
@@ -387,14 +438,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);
 
@@ -505,6 +550,7 @@ export class WorkflowTriggerActivator {
 		nodeIds: Set<INode['id']>,
 		outcome: TriggerActivationOutcome,
 		activationMode: WorkflowActivateMode,
+		abort: TriggerOperationAbort,
 	) {
 		const webhooksByNode = this.groupWebhookTriggersByNode(workflow, additionalData, nodeIds);
 
@@ -516,6 +562,7 @@ export class WorkflowTriggerActivator {
 					nodeName,
 					webhooks,
 					activationMode,
+					abort,
 				),
 		);
 
@@ -536,18 +583,23 @@ export class WorkflowTriggerActivator {
 		nodeName: string,
 		webhooks: IWebhookData[],
 		activationMode: WorkflowActivateMode,
+		abort: TriggerOperationAbort,
 	): Promise<Result<{ nodeId: INode['id'] }, TriggerActivationFailure>> {
 		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 });
@@ -588,8 +640,10 @@ export class WorkflowTriggerActivator {
 		workflow: Workflow,
 		additionalData: IWorkflowExecuteAdditionalData,
 		nodeIds: Set<INode['id']>,
+		abort: TriggerOperationAbort,
 	) {
 		const removedNodeNames: string[] = [];
+		let firstFailure: Error | undefined;
 
 		await workflow.expression.acquireIsolate();
 		try {
@@ -598,18 +652,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<string, number>();
+			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;
@@ -640,6 +719,7 @@ export class WorkflowTriggerActivator {
 		nodeIds: Set<INode['id']>,
 		outcome: TriggerActivationOutcome,
 		activationMode: WorkflowActivateMode,
+		abort: TriggerOperationAbort,
 	) {
 		const triggerNodeIds = this.getNonWebhookTriggerNodeIdsForNodeIds(workflow, nodeIds);
 		if (triggerNodeIds.length === 0) return;
@@ -665,10 +745,16 @@ export class WorkflowTriggerActivator {
 		const results = await Promise.all(
 			triggerNodeIds.map(async (nodeId): Promise<Result<INode['id'], TriggerActivationFailure>> => {
 				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);
@@ -703,12 +789,14 @@ export class WorkflowTriggerActivator {
 		workflowId: WorkflowId,
 		workflow: Workflow,
 		nodeIds: Set<INode['id']>,
+		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),
 			),
 		);
 	}
@@ -772,14 +860,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);
 

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(

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();