refactor(core): Run publication outbox consumers as a capped pool of independent workers (no-changelog) (#36303)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Mike Repeć <mike.repec@n8n.io>

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

@@ -179,37 +179,42 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 	});
 
 	describe('concurrent drains', () => {
-		test('coalesces overlapping drainPending calls onto a single pass', async () => {
+		test('overlapping drainPending calls never exceed the concurrency cap', async () => {
 			const record = makeRecord({ id: 1 });
-			let releaseClaim!: () => void;
-			const claimGate = new Promise<void>((resolve) => {
-				releaseClaim = resolve;
+			let releaseApply!: () => void;
+			let signalApplyStarted!: () => void;
+			const applyStarted = new Promise<void>((resolve) => {
+				signalApplyStarted = resolve;
+			});
+			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);
+			applier.apply.mockImplementationOnce(async () => {
+				signalApplyStarted();
+				await new Promise<void>((resolve) => {
+					releaseApply = resolve;
+				});
+				return { type: 'completed', triggerStatuses: [] };
 			});
-			outboxRepository.claimNextPendingRecord
-				.mockImplementationOnce(async () => {
-					await claimGate;
-					return record;
-				})
-				.mockResolvedValue(null);
 			consumer.startPolling();
 
 			const first = consumer.drainPending();
+			await applyStarted;
 			const second = consumer.drainPending();
+			await vi.advanceTimersByTimeAsync(0);
 
-			releaseClaim();
-			const [firstProcessed, secondProcessed] = await Promise.all([first, second]);
+			// The cap is one worker, so the second call must not start another claim
+			// while the first worker is busy.
+			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);
+
+			releaseApply();
+			await Promise.all([first, second]);
 
-			// Both callers share the one in-flight pass, so the record is claimed and
-			// applied exactly once even though drainPending was invoked twice.
 			expect(applier.apply).toHaveBeenCalledTimes(1);
 			expect(reporter.report).toHaveBeenCalledTimes(1);
-			expect(firstProcessed).toBe(1);
-			expect(secondProcessed).toBe(1);
 		});
 	});
 
 	describe('parallel drains', () => {
-		test('processes up to the configured concurrency in parallel and returns the total', async () => {
+		test('processes up to the configured concurrency in parallel', async () => {
 			consumer = createConsumer(true, true, 2);
 
 			// Distinct workflowIds so the per-workflow lifecycle lock never serializes them.
@@ -252,8 +257,129 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 
 			releases.get(2)!();
 			releases.get(3)!();
-			const processed = await drain;
-			expect(processed).toBe(3);
+			await drain;
+			expect(reporter.report).toHaveBeenCalledTimes(3);
+		});
+	});
+
+	describe('worker pool', () => {
+		test('a wake-up processes new records while another worker is stuck on a hung record', async () => {
+			consumer = createConsumer(true, true, 2);
+			const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' });
+			let signalStuckStarted!: () => void;
+			const stuckStarted = new Promise<void>((resolve) => {
+				signalStuckStarted = resolve;
+			});
+			applier.apply.mockImplementation(async (record) => {
+				if (record.workflowId === 'wf-stuck') {
+					signalStuckStarted();
+					return await new Promise(() => {});
+				}
+				return { type: 'completed', triggerStatuses: [] };
+			});
+			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
+			consumer.startPolling();
+
+			void consumer.drainPending();
+			await stuckStarted;
+
+			// A new record commits and its wake-up arrives while the first record hangs.
+			const fresh = makeRecord({ id: 2, workflowId: 'wf-2' });
+			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(fresh).mockResolvedValue(null);
+			void consumer.drainPending();
+			await vi.advanceTimersByTimeAsync(0);
+
+			expect(applier.apply).toHaveBeenCalledWith(fresh, expect.anything());
+			expect(reporter.report).toHaveBeenCalledWith(
+				fresh,
+				expect.objectContaining({ type: 'completed' }),
+			);
+		});
+
+		test('a wake-up arriving while all workers are busy triggers a follow-up pass', async () => {
+			const r1 = makeRecord({ id: 1, workflowId: 'wf-1' });
+			const r2 = makeRecord({ id: 2, workflowId: 'wf-2' });
+			outboxRepository.claimNextPendingRecord
+				.mockResolvedValueOnce(r1)
+				// The pass runs dry before r2 commits, mirroring a record that lands
+				// after the final claim's visibility horizon.
+				.mockResolvedValueOnce(null)
+				.mockResolvedValueOnce(r2)
+				.mockResolvedValue(null);
+			let releaseApply!: () => void;
+			let signalApplyStarted!: () => void;
+			const applyStarted = new Promise<void>((resolve) => {
+				signalApplyStarted = resolve;
+			});
+			applier.apply.mockImplementationOnce(async () => {
+				signalApplyStarted();
+				await new Promise<void>((resolve) => {
+					releaseApply = resolve;
+				});
+				return { type: 'completed', triggerStatuses: [] };
+			});
+			consumer.startPolling();
+
+			const first = consumer.drainPending();
+			await applyStarted;
+
+			// r2's wake-up arrives while the only worker is still busy.
+			const second = consumer.drainPending();
+			releaseApply();
+			await Promise.all([first, second]);
+
+			expect(applier.apply).toHaveBeenCalledTimes(2);
+			expect(applier.apply).toHaveBeenCalledWith(r2, expect.anything());
+		});
+
+		test('drainPending rejects with a wrapping error after the pool reported the failure', async () => {
+			const error = new Error('claim failed');
+			outboxRepository.claimNextPendingRecord.mockRejectedValueOnce(error);
+			consumer.startPolling();
+
+			// Awaiting callers (e.g. the reconciler) must still observe drain
+			// failures, or they would report success for records nobody claimed.
+			// The wrapper lets them fail their own operation; the underlying error
+			// is reported by the pool alone, exactly once. `shouldReport: false`
+			// pins that a caller's generic catch reporting the wrapper creates no
+			// second Sentry event (the ErrorReporter drops non-reportable errors).
+			await expect(consumer.drainPending()).rejects.toMatchObject({
+				message: expect.stringContaining('drain failed'),
+				cause: error,
+				shouldReport: false,
+			});
+			expect(errorReporter.error).toHaveBeenCalledTimes(1);
+			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
+		});
+
+		test('an idle pass starts no tracing span', async () => {
+			consumer.startPolling();
+
+			await consumer.drainPending();
+
+			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);
+			expect(tracing.startSpan).not.toHaveBeenCalled();
+		});
+
+		test('the poll fallback keeps running while a worker is stuck on a hung record', async () => {
+			consumer = createConsumer(true, true, 2);
+			const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' });
+			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
+			applier.apply.mockImplementationOnce(async () => await new Promise(() => {}));
+			consumer.startPolling();
+
+			// First cycle claims the record that hangs.
+			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);
+
+			const fresh = makeRecord({ id: 2, workflowId: 'wf-2' });
+			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(fresh).mockResolvedValue(null);
+			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);
+
+			expect(applier.apply).toHaveBeenCalledWith(fresh, expect.anything());
+			expect(reporter.report).toHaveBeenCalledWith(
+				fresh,
+				expect.objectContaining({ type: 'completed' }),
+			);
 		});
 	});
 
@@ -361,7 +487,7 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 
 			// 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);
+			await drain;
 			expect(reporter.report).not.toHaveBeenCalled();
 			expect(errorReporter.error).toHaveBeenCalledWith(
 				expect.objectContaining({
@@ -373,8 +499,12 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 			// 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);
+			await consumer.drainPending();
 			expect(reporter.report).toHaveBeenCalledTimes(1);
+			expect(reporter.report).toHaveBeenCalledWith(
+				next,
+				expect.objectContaining({ type: 'completed' }),
+			);
 		});
 
 		test('a record that settles during the grace period still gets its terminal status', async () => {
@@ -395,7 +525,7 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 			const drain = consumer.drainPending();
 			await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS + 2000);
 
-			await expect(drain).resolves.toBe(1);
+			await drain;
 			expect(reporter.report).toHaveBeenCalledWith(
 				record,
 				expect.objectContaining({ type: 'completed' }),
@@ -416,7 +546,7 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 			const drain = consumer.drainPending();
 			await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS);
 
-			await expect(drain).resolves.toBe(1);
+			await drain;
 			expect(reporter.report).toHaveBeenCalledWith(
 				record,
 				expect.objectContaining({
@@ -482,7 +612,7 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 			const drain = consumer.drainPending();
 			await vi.advanceTimersByTimeAsync(15_200);
 
-			await expect(drain).resolves.toBe(0);
+			await drain;
 			expect(errorReporter.error).toHaveBeenCalledWith(
 				expect.objectContaining({
 					message: expect.stringContaining('Abandoned workflow publication outbox record'),
@@ -577,6 +707,35 @@ describe('WorkflowPublicationOutboxConsumer', () => {
 			expect(outboxRepository.claimNextPendingRecord).not.toHaveBeenCalled();
 			expect(vi.getTimerCount()).toBe(0);
 		});
+
+		test('never rejects on a drain failure, which is reported exactly once', async () => {
+			const error = new Error('claim failed');
+			outboxRepository.claimNextPendingRecord.mockRejectedValueOnce(error);
+
+			// Pubsub dispatch drops handler rejections, so a rejecting wakeUp would
+			// surface as an unhandled promise rejection.
+			await expect(consumer.wakeUp()).resolves.toBeUndefined();
+
+			expect(errorReporter.error).toHaveBeenCalledTimes(1);
+			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
+		});
+
+		test('overlapping wake-ups report a shared drain failure exactly once', async () => {
+			const error = new Error('claim failed');
+			let rejectClaim!: (reason: Error) => void;
+			const claimGate = new Promise<WorkflowPublicationOutbox | null>((_resolve, reject) => {
+				rejectClaim = reject;
+			});
+			outboxRepository.claimNextPendingRecord.mockImplementationOnce(async () => await claimGate);
+
+			const first = consumer.wakeUp();
+			const second = consumer.wakeUp();
+			rejectClaim(error);
+			await Promise.all([first, second]);
+
+			expect(errorReporter.error).toHaveBeenCalledTimes(1);
+			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
+		});
 	});
 
 	describe('poll cycle', () => {

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

@@ -0,0 +1,132 @@
+import { WorkflowPublicationOutboxWorkerPool } from '@/workflows/publication/workflow-publication-outbox-worker-pool';
+
+describe('WorkflowPublicationOutboxWorkerPool', () => {
+	const deferred = () => {
+		let resolve!: () => void;
+		const promise = new Promise<void>((r) => (resolve = r));
+		return { promise, resolve };
+	};
+
+	const createPool = ({
+		runPass = async () => {},
+		shouldRun = () => true,
+		concurrency = 2,
+		onWorkerError = vi.fn(),
+	}: Partial<{
+		runPass: () => Promise<void>;
+		shouldRun: () => boolean;
+		concurrency: number;
+		onWorkerError: (error: Error) => void;
+	}> = {}) =>
+		new WorkflowPublicationOutboxWorkerPool({ runPass, shouldRun, concurrency, onWorkerError });
+
+	test('topUp spawns exactly `concurrency` workers', async () => {
+		const runPass = vi.fn(async () => {});
+		const pool = createPool({ runPass, concurrency: 3 });
+
+		pool.topUp();
+		await pool.awaitIdle();
+
+		expect(runPass).toHaveBeenCalledTimes(3);
+	});
+
+	test('topUp is a no-op when shouldRun is false', async () => {
+		const runPass = vi.fn(async () => {});
+		const pool = createPool({ runPass, shouldRun: () => false });
+
+		pool.topUp();
+
+		await expect(pool.awaitIdle()).resolves.toBeNull();
+		expect(runPass).not.toHaveBeenCalled();
+	});
+
+	test('topUp at capacity spawns no extra worker but flags a follow-up pass', async () => {
+		const gate = deferred();
+		const runPass = vi.fn(async () => await gate.promise);
+		const pool = createPool({ runPass, concurrency: 1 });
+
+		pool.topUp();
+		pool.topUp(); // at capacity: must only flag a follow-up
+		expect(runPass).toHaveBeenCalledTimes(1);
+
+		gate.resolve();
+		await pool.awaitIdle();
+
+		expect(runPass).toHaveBeenCalledTimes(2);
+	});
+
+	test('no follow-up pass when shouldRun turns false before the worker exits', async () => {
+		const gate = deferred();
+		let running = true;
+		const runPass = vi.fn(async () => await gate.promise);
+		const pool = createPool({ runPass, shouldRun: () => running, concurrency: 1 });
+
+		pool.topUp();
+		pool.topUp();
+		running = false;
+		gate.resolve();
+		await pool.awaitIdle();
+
+		expect(runPass).toHaveBeenCalledTimes(1);
+	});
+
+	test('awaitIdle resolves null when every pass completes', async () => {
+		const pool = createPool();
+
+		pool.topUp();
+
+		await expect(pool.awaitIdle()).resolves.toBeNull();
+	});
+
+	test('a failed pass never rejects: its error is handed to onWorkerError once and returned by awaitIdle', async () => {
+		const error = new Error('pass failed');
+		const onWorkerError = vi.fn();
+		const pool = createPool({
+			runPass: async () => await Promise.reject(error),
+			concurrency: 1,
+			onWorkerError,
+		});
+
+		pool.topUp();
+
+		await expect(pool.awaitIdle()).resolves.toBe(error);
+		expect(onWorkerError).toHaveBeenCalledTimes(1);
+		expect(onWorkerError).toHaveBeenCalledWith(error);
+	});
+
+	test('one failed pass does not affect the other workers', async () => {
+		const gate = deferred();
+		const onWorkerError = vi.fn();
+		let calls = 0;
+		const runPass = vi.fn(async () => {
+			if (++calls === 1) throw new Error('first pass failed');
+			await gate.promise;
+		});
+		const pool = createPool({ runPass, concurrency: 2, onWorkerError });
+
+		pool.topUp();
+		gate.resolve();
+		const error = await pool.awaitIdle();
+
+		expect(error?.message).toBe('first pass failed');
+		expect(runPass).toHaveBeenCalledTimes(2);
+		expect(onWorkerError).toHaveBeenCalledTimes(1);
+	});
+
+	test('awaitIdle waits for a follow-up pass spawned after it started observing', async () => {
+		const gate = deferred();
+		let calls = 0;
+		const runPass = vi.fn(async () => {
+			if (++calls === 1) await gate.promise;
+		});
+		const pool = createPool({ runPass, concurrency: 1 });
+
+		pool.topUp();
+		const idle = pool.awaitIdle();
+		pool.topUp(); // flags the follow-up while the first worker is busy
+		gate.resolve();
+		await idle;
+
+		expect(runPass).toHaveBeenCalledTimes(2);
+	});
+});

packages/cli/src/workflows/publication/__tests__/workflow-publication-reconciler.service.test.ts

@@ -66,7 +66,7 @@ beforeEach(() => {
 	outboxRepository.findVersionSkewedWorkflowIds.mockResolvedValue([]);
 	outboxRepository.findTriggerStatusDriftedWorkflowIds.mockResolvedValue([]);
 	outboxRepository.findUnreportedPublishedWorkflowIds.mockResolvedValue([]);
-	outboxConsumer.drainPending.mockResolvedValue(0);
+	outboxConsumer.drainPending.mockResolvedValue();
 	setRegistered({});
 	activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds.mockReturnValue([]);
 	activeWorkflowTriggers.remove.mockResolvedValue(true);

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

@@ -17,6 +17,7 @@ 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 { WorkflowPublicationOutboxWorkerPool } from '@/workflows/publication/workflow-publication-outbox-worker-pool';
 import type { TriggerOperationAbort } from '@/workflows/triggers/workflow-trigger-activator';
 
 /**
@@ -42,12 +43,18 @@ const ABANDON_GRACE_LEASE_FRACTION = 0.25;
 
 /**
  * Consumes the workflow publication outbox on the leader instance. It owns the
- * queue mechanics only: the poll loop and claiming the next pending record. For
- * each claimed record it delegates to the applier (which reconciles triggers
- * and returns a result) and then to the reporter (which writes the terminal
- * status and side effects). Any unexpected error from the applier is turned
- * into a failed result so the reporter remains the single writer of terminal
- * outbox statuses.
+ * queue mechanics only: the poll loop and a pool of up to
+ * `workflowPublicationConcurrency` independent workers, each claiming pending
+ * records until none remain. For each claimed record a worker delegates to the
+ * applier (which reconciles triggers and returns a result) and then to the
+ * reporter (which writes the terminal status and side effects). Any unexpected
+ * error from the applier is turned into a failed result so the reporter
+ * remains the single writer of terminal outbox statuses.
+ *
+ * Coordination lives in the database, not in this class: claiming is atomic
+ * and per-workflow serialized, so workers (even across passes) never
+ * double-process a record. A worker stuck on one record therefore only
+ * occupies its own pool slot — the remaining slots keep serving new records.
  */
 @Service()
 export class WorkflowPublicationOutboxConsumer {
@@ -57,7 +64,7 @@ export class WorkflowPublicationOutboxConsumer {
 
 	private isShuttingDown = false;
 
-	private activeDrain: Promise<number> | null = null;
+	private readonly pool: WorkflowPublicationOutboxWorkerPool;
 
 	constructor(
 		private readonly logger: Logger,
@@ -72,6 +79,12 @@ export class WorkflowPublicationOutboxConsumer {
 		private readonly eventService: EventService,
 	) {
 		this.logger = this.logger.scoped('workflow-publication');
+		this.pool = new WorkflowPublicationOutboxWorkerPool({
+			runPass: async () => await this.runWorkerPass(),
+			shouldRun: () => this.shouldKeepPolling(),
+			concurrency: this.workflowsConfig.workflowPublicationConcurrency,
+			onWorkerError: (error) => this.errorReporter.error(error, { shouldBeLogged: true }),
+		});
 	}
 
 	async init() {
@@ -105,8 +118,9 @@ export class WorkflowPublicationOutboxConsumer {
 	async shutdown() {
 		this.isShuttingDown = true;
 		this.stopPolling();
-		// Wait for any in-flight drain to finish so triggers aren't left half-activated.
-		await this.activeDrain;
+		// Wait for in-flight workers to finish their current record so triggers
+		// aren't left half-activated.
+		await this.pool.awaitIdle();
 	}
 
 	/**
@@ -125,94 +139,102 @@ export class WorkflowPublicationOutboxConsumer {
 		if (!this.instanceSettings.isLeader) return;
 
 		this.startPolling();
-		await this.drainPending();
+		try {
+			await this.drainPending();
+		} catch (error) {
+			// The pool already reported the underlying failure, and pubsub dispatch
+			// drops handler rejections (they would surface as unhandled promise
+			// rejections) — so neither rethrow nor re-report, just trace.
+			this.logger.debug('Publication outbox drain triggered by a wake-up failed', {
+				error: ensureError(error).message,
+			});
+		}
 	}
 
 	// The `workflow-publish-wake-up` event drains the outbox promptly (see
 	// `wakeUp`); the poller is kept as a fallback since pubsub delivery is not
-	// ensured.
+	// ensured. Topping up the pool is synchronous, so the next cycle is armed
+	// immediately — a worker stuck on a record never wedges the poll loop.
 	private schedulePollCycle() {
 		clearTimeout(this.pollTimeout);
 		if (!this.shouldKeepPolling()) return;
 
-		this.pollTimeout = setTimeout(async () => {
-			try {
-				await this.pollCycle();
-			} catch (error) {
-				this.errorReporter.error(error, { shouldBeLogged: true });
-			}
-
+		this.pollTimeout = setTimeout(() => {
+			this.pool.topUp();
 			if (this.shouldKeepPolling()) this.schedulePollCycle();
 		}, this.workflowsConfig.publicationOutboxPollIntervalMs);
 	}
 
-	private async pollCycle() {
-		const processed = await this.drainPending();
-
-		// Only log if we processed more than 1 since we log each individual record
-		if (processed > 1) {
-			this.logger.debug(`Processed ${processed} workflow publication outbox records in this cycle`);
+	/**
+	 * Top the worker pool up to the configured concurrency and resolve once the
+	 * pool goes idle. Used by the pubsub wake-up, by leader startup for an
+	 * immediate drain, and by the reconciler after enqueueing records. Workers
+	 * stop claiming if the instance steps down or shuts down mid-pass.
+	 *
+	 * Resolution waits for every active worker, so a worker stuck on a hung
+	 * record delays it (bounded by the abort/abandon deadline) — but not the
+	 * processing of other records, which the remaining slots keep serving.
+	 * Rejects once the pool idles if a worker pass failed, with an error
+	 * wrapping one of the failures. The pool has already reported the underlying
+	 * error (exactly once, at the source); the rejection exists so awaiting
+	 * callers can fail their own operation (e.g. the reconciler's failure
+	 * telemetry) — they report their layer's outcome, never the pool's error.
+	 */
+	async drainPending(): Promise<void> {
+		this.pool.topUp();
+		const error = await this.pool.awaitIdle();
+		if (error) {
+			// `level: 'warning'` makes the wrapper non-reportable (`shouldReport`
+			// false), so an awaiting caller's generic catch passing it to the
+			// ErrorReporter creates no second Sentry event — the pool's source
+			// report of the cause stays the single report of the failure.
+			throw new OperationalError('Workflow publication outbox drain failed', {
+				cause: error,
+				level: 'warning',
+			});
 		}
 	}
 
 	/**
-	 * Claim and process every currently pending record in a single pass, returning
-	 * the number processed. Used both by the scheduled poll cycle and at leader
-	 * startup for an immediate drain. The loop stops if the instance steps down or
-	 * shuts down mid-drain; claiming is atomic, so an extra concurrent drain never
-	 * double-processes a record. Concurrent callers coalesce onto the same in-flight
-	 * pass rather than running an overlapping drain, which also lets shutdown wait
-	 * for the active pass to settle.
+	 * Claim and process pending records until none remain. An idle pass — the
+	 * first claim finds nothing — starts no tracing span, so an idle leader
+	 * doesn't emit one empty span per worker per poll tick.
 	 */
-	async drainPending(): Promise<number> {
-		if (this.activeDrain) return await this.activeDrain;
+	private async runWorkerPass(): Promise<void> {
+		if (!this.shouldKeepPolling()) return;
 
-		const drain = this.runDrain();
-		this.activeDrain = drain;
-		try {
-			return await drain;
-		} finally {
-			this.activeDrain = null;
-		}
-	}
+		const first = await this.outboxRepository.claimNextPendingRecord();
+		if (!first) return;
 
-	private async runDrain(): Promise<number> {
-		const concurrency = this.workflowsConfig.workflowPublicationConcurrency;
-		return await this.tracing.startSpan(
+		await this.tracing.startSpan(
 			{
 				name: 'Publication outbox drain',
 				op: 'publication.outbox.drain',
-				attributes: { 'n8n.publication.consumer_concurrency': concurrency },
+				attributes: {
+					'n8n.publication.consumer_concurrency':
+						this.workflowsConfig.workflowPublicationConcurrency,
+				},
 			},
 			async (span) => {
 				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 workerFailed = false;
-				const runWorker = async () => {
-					while (!workerFailed && this.shouldKeepPolling()) {
-						const record = await this.outboxRepository.claimNextPendingRecord();
-						if (!record) break;
+				let record: WorkflowPublicationOutbox | null = first;
+				while (record) {
+					if (await this.processRecordWithAbort(record)) processed++;
 
-						if (await this.processRecordWithAbort(record)) processed++;
-					}
-				};
+					record = this.shouldKeepPolling()
+						? await this.outboxRepository.claimNextPendingRecord()
+						: null;
+				}
 
-				const workerTasks = Array.from({ length: concurrency }, async () => {
-					await runWorker().catch((error) => {
-						workerFailed = true;
-						throw error;
-					});
-				});
-
-				const results = await Promise.allSettled(workerTasks);
-
-				const failure = results.find((r) => r.status === 'rejected');
-				if (failure?.status === 'rejected') throw failure.reason;
+				// Only log if we processed more than 1 since we log each individual record
+				if (processed > 1) {
+					this.logger.debug(
+						`Processed ${processed} workflow publication outbox records in this pass`,
+					);
+				}
 
 				span.setAttribute('n8n.publication.records_processed', processed);
 				span.setStatus({ code: SpanStatus.ok });
-				return processed;
 			},
 		);
 	}

packages/cli/src/workflows/publication/workflow-publication-outbox-worker-pool.ts

@@ -0,0 +1,87 @@
+import { ensureError } from '@n8n/utils/errors/ensure-error';
+
+interface OutboxWorkerPoolOptions {
+	/** A single worker's work: claim and process records until none remain. */
+	runPass: () => Promise<void>;
+	/** Whether the pool may (keep) spawning workers, checked before every spawn. */
+	shouldRun: () => boolean;
+	/** Maximum number of concurrent workers. */
+	concurrency: number;
+	/** The owner's error policy, invoked exactly once per failed pass, at the source. */
+	onWorkerError: (error: Error) => void;
+}
+
+/**
+ * A capped pool of independent, concurrent runs of `runPass`. Owns the pool
+ * mechanics only: topping up to `concurrency` workers, containing each
+ * pass's failure to its own slot, the follow-up pass for a wake-up that
+ * arrives at capacity, and awaiting idleness. Everything domain-specific
+ * belongs to the owner, injected via {@link OutboxWorkerPoolOptions}.
+ */
+export class WorkflowPublicationOutboxWorkerPool {
+	/** Worker passes in flight. A worker never rejects (so `Promise.all` over
+	 * the set is safe): it settles with the error that ended its pass, or
+	 * `null` when the pass completed. */
+	private readonly activeWorkers = new Set<Promise<Error | null>>();
+
+	/** Set when a wake-up arrives while the pool is at capacity; a worker
+	 * exiting then starts a follow-up pass, so a record committed after the
+	 * running workers' final claims is still picked up promptly. */
+	private wakeRequested = false;
+
+	constructor(private readonly options: OutboxWorkerPoolOptions) {}
+
+	/**
+	 * Top the pool up to the configured concurrency, or flag a follow-up pass
+	 * when already at capacity. Synchronous, so a caller (e.g. the poll loop)
+	 * is never wedged by a stuck worker.
+	 */
+	topUp() {
+		if (!this.options.shouldRun()) return;
+
+		const { concurrency } = this.options;
+		if (this.activeWorkers.size >= concurrency) {
+			// The running workers' final claims may predate the record behind this
+			// wake-up; have the next worker to exit run a follow-up pass.
+			this.wakeRequested = true;
+			return;
+		}
+
+		while (this.activeWorkers.size < concurrency) this.spawnWorker();
+	}
+
+	/** Resolves once the pool is idle, with one of the worker errors if any
+	 * pass failed (which one is arbitrary; every failure is already handed to
+	 * `onWorkerError` at the source). */
+	async awaitIdle(): Promise<Error | null> {
+		let error: Error | null = null;
+		// Loop: a follow-up pass (see `wakeRequested`) can repopulate the pool
+		// after the currently observed workers settle.
+		while (this.activeWorkers.size > 0) {
+			const outcomes = await Promise.all([...this.activeWorkers]);
+			error ??= outcomes.find((outcome) => outcome !== null) ?? null;
+		}
+		return error;
+	}
+
+	private spawnWorker() {
+		const worker = this.options
+			.runPass()
+			.then(
+				() => null,
+				(error) => {
+					const failure = ensureError(error);
+					this.options.onWorkerError(failure);
+					return failure;
+				},
+			)
+			.finally(() => {
+				this.activeWorkers.delete(worker);
+				if (this.wakeRequested) {
+					this.wakeRequested = false;
+					if (this.options.shouldRun()) this.spawnWorker();
+				}
+			});
+		this.activeWorkers.add(worker);
+	}
+}