From f00a7044e8b683d7caa2d6a9d0b7f7c614fb0ddb Mon Sep 17 00:00:00 2001 From: mfsiega <93014743+mfsiega@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:14:18 +0000 Subject: [PATCH] refactor(core): Run publication outbox consumers as a capped pool of independent workers (no-changelog) (#36303) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Fable 5 Co-authored-by: Mike Repeć --- ...rkflow-publication-outbox-consumer.test.ts | 207 ++++++++++++++++-- ...low-publication-outbox-worker-pool.test.ts | 132 +++++++++++ ...low-publication-reconciler.service.test.ts | 2 +- .../workflow-publication-outbox-consumer.ts | 156 +++++++------ ...workflow-publication-outbox-worker-pool.ts | 87 ++++++++ 5 files changed, 492 insertions(+), 92 deletions(-) create mode 100644 packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-worker-pool.test.ts create mode 100644 packages/cli/src/workflows/publication/workflow-publication-outbox-worker-pool.ts diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts index 0dcb27044de..9f3030c7645 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-consumer.test.ts @@ -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((resolve) => { - releaseClaim = resolve; + let releaseApply!: () => void; + let signalApplyStarted!: () => void; + const applyStarted = new Promise((resolve) => { + signalApplyStarted = resolve; + }); + outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null); + applier.apply.mockImplementationOnce(async () => { + signalApplyStarted(); + await new Promise((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((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((resolve) => { + signalApplyStarted = resolve; + }); + applier.apply.mockImplementationOnce(async () => { + signalApplyStarted(); + await new Promise((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((_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', () => { diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-worker-pool.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-outbox-worker-pool.test.ts new file mode 100644 index 00000000000..f7f37e17e83 --- /dev/null +++ b/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((r) => (resolve = r)); + return { promise, resolve }; + }; + + const createPool = ({ + runPass = async () => {}, + shouldRun = () => true, + concurrency = 2, + onWorkerError = vi.fn(), + }: Partial<{ + runPass: () => Promise; + 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); + }); +}); diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-reconciler.service.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-reconciler.service.test.ts index c5e0e1cdb34..f7187203007 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-reconciler.service.test.ts +++ b/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); diff --git a/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts b/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts index eb2624942c5..4c799024d7d 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts +++ b/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 | 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 { + 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 { - if (this.activeDrain) return await this.activeDrain; + private async runWorkerPass(): Promise { + 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 { - 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; }, ); } diff --git a/packages/cli/src/workflows/publication/workflow-publication-outbox-worker-pool.ts b/packages/cli/src/workflows/publication/workflow-publication-outbox-worker-pool.ts new file mode 100644 index 00000000000..824f4d8063a --- /dev/null +++ b/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; + /** 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>(); + + /** 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 { + 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); + } +}