From 1801f65d85534ccfc9945cf78b159e314e6a16b1 Mon Sep 17 00:00:00 2001 From: Danny Martini Date: Tue, 4 Aug 2026 08:24:49 +0200 Subject: [PATCH] feat(core): Converge leader takeover through the reconciliation loop (#35454) Co-authored-by: Claude Fable 5 --- .../workflow-publication-outbox.repository.ts | 100 +++--------- packages/cli/src/commands/start.ts | 9 -- .../published-workflow-enqueuer.test.ts | 51 ------ ...ished-workflow-trigger-deactivator.test.ts | 65 -------- ...low-publication-reconciler.service.test.ts | 108 ++++++++++--- .../published-workflow-enqueuer.ts | 48 ------ .../published-workflow-trigger-deactivator.ts | 31 +--- .../workflow-publication-outbox-consumer.ts | 3 +- ...workflow-publication-reconciler.service.ts | 95 ++++++----- .../integration/commands/start.cmd.test.ts | 2 +- ...flow-publication-outbox.repository.test.ts | 151 ++---------------- ...rkflow-publication-outbox-consumer.test.ts | 23 --- .../workflow-publication-reconciler.test.ts | 11 +- 13 files changed, 193 insertions(+), 504 deletions(-) delete mode 100644 packages/cli/src/workflows/publication/__tests__/published-workflow-enqueuer.test.ts delete mode 100644 packages/cli/src/workflows/publication/published-workflow-enqueuer.ts diff --git a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts index 8fe3b8b1b30..c74e7f808aa 100644 --- a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts @@ -11,6 +11,9 @@ import { } from '../entities/workflow-publication-outbox'; import { isUniqueConstraintError } from '../utils/is-unique-constraint-error'; +/** Sqlite bound-variable budget per statement; safely under every build's cap. */ +const SQLITE_ENQUEUE_CHUNK_SIZE = 999; + @Service() export class WorkflowPublicationOutboxRepository extends Repository { constructor( @@ -92,75 +95,6 @@ export class WorkflowPublicationOutboxRepository extends Repository { - if (this.globalConfig.database.type === 'postgresdb') { - await this.enqueueForLeaderHandoffWithPostgresUpsert(); - return; - } - - await this.enqueueForLeaderHandoffWithSqliteUpsert(); - } - - private async enqueueForLeaderHandoffWithPostgresUpsert(): Promise { - const outboxTableName = this.getTableName('workflow_publication_outbox'); - const workflowTableName = this.getTableName('workflow_entity'); - const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status'); - - await this.query( - `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status") - SELECT w."id", w."activeVersionId", '${Status.Pending}' - FROM ${workflowTableName} w - WHERE w."activeVersionId" IS NOT NULL AND w."isArchived" = false - AND ( - -- Enqueue workflows with at least one in-memory trigger - EXISTS ( - SELECT 1 FROM ${triggerStatusTableName} ts - WHERE ts."workflowId" = w."id" AND ts."triggerKind" = 'in-memory' - ) - -- Enqueue workflows where we have no trigger status data - OR NOT EXISTS (SELECT 1 FROM ${triggerStatusTableName} ts WHERE ts."workflowId" = w."id") - ) - ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}') - DO UPDATE SET "publishedVersionId" = EXCLUDED."publishedVersionId", "updatedAt" = CURRENT_TIMESTAMP(3)`, - ); - } - - private async enqueueForLeaderHandoffWithSqliteUpsert(): Promise { - const outboxTableName = this.getTableName('workflow_publication_outbox'); - const workflowTableName = this.getTableName('workflow_entity'); - const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status'); - - await this.query( - `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status") - SELECT w."id", w."activeVersionId", '${Status.Pending}' - FROM ${workflowTableName} w - WHERE w."activeVersionId" IS NOT NULL AND w."isArchived" = 0 - AND ( - -- Enqueue workflows with at least one in-memory trigger - EXISTS ( - SELECT 1 FROM ${triggerStatusTableName} ts - WHERE ts."workflowId" = w."id" AND ts."triggerKind" = 'in-memory' - ) - -- Enqueue workflows where we have no trigger status data - OR NOT EXISTS (SELECT 1 FROM ${triggerStatusTableName} ts WHERE ts."workflowId" = w."id") - ) - ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}') - DO UPDATE SET "publishedVersionId" = excluded."publishedVersionId", "updatedAt" = STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')`, - ); - } - /** * Enqueue a pending publication record for each given workflow that still * exists, in a single statement. Used by reconciliation, which must be able to @@ -210,17 +144,25 @@ export class WorkflowPublicationOutboxRepository extends Repository { const outboxTableName = this.getTableName('workflow_publication_outbox'); const workflowTableName = this.getTableName('workflow_entity'); - const placeholders = workflowIds.map(() => '?').join(', '); - await this.query( - `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status") - SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}' - FROM ${workflowTableName} w - WHERE w."id" IN (${placeholders}) - ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}') - DO NOTHING`, - workflowIds, - ); + // One placeholder is bound per id and sqlite caps bound variables per + // statement, while a fresh leader can pass every active workflow at once. + // Chunked upserts stay idempotent, so a failure mid-batch just leaves the + // remainder for the next reconciliation pass. + for (let i = 0; i < workflowIds.length; i += SQLITE_ENQUEUE_CHUNK_SIZE) { + const chunk = workflowIds.slice(i, i + SQLITE_ENQUEUE_CHUNK_SIZE); + const placeholders = chunk.map(() => '?').join(', '); + + await this.query( + `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status") + SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}' + FROM ${workflowTableName} w + WHERE w."id" IN (${placeholders}) + ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}') + DO NOTHING`, + chunk, + ); + } } /** diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 137fa69dac7..8da65b9f20e 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -417,9 +417,6 @@ export class Start extends BaseCommand> { // Start to get active workflows and run their triggers if (this.globalConfig.workflows.useWorkflowPublicationService) { - const { PublishedWorkflowEnqueuer } = await import( - '@/workflows/publication/published-workflow-enqueuer.js' - ); const { WorkflowPublicationOutboxConsumer } = await import( '@/workflows/publication/workflow-publication-outbox-consumer.js' ); @@ -439,12 +436,6 @@ export class Start extends BaseCommand> { // listeners, so rewire to pick them up. Container.get(PubSubRegistry).init(); - // Enqueue needs to happen before outbox consumer init, so it can activate - // everything on the first drain - if (this.instanceSettings.isLeader) { - await Container.get(PublishedWorkflowEnqueuer).enqueueActiveWorkflows(); - } - // Don't await: the immediate drain activates every trigger and can take a // while, so let it run in the background instead of blocking startup. void Container.get(WorkflowPublicationOutboxConsumer) diff --git a/packages/cli/src/workflows/publication/__tests__/published-workflow-enqueuer.test.ts b/packages/cli/src/workflows/publication/__tests__/published-workflow-enqueuer.test.ts deleted file mode 100644 index e3e396053f0..00000000000 --- a/packages/cli/src/workflows/publication/__tests__/published-workflow-enqueuer.test.ts +++ /dev/null @@ -1,51 +0,0 @@ -import type { Logger } from '@n8n/backend-common'; -import type { WorkflowsConfig } from '@n8n/config'; -import type { WorkflowPublicationOutboxRepository } from '@n8n/db'; -import { mock } from 'vitest-mock-extended'; - -import { PublishedWorkflowEnqueuer } from '@/workflows/publication/published-workflow-enqueuer'; -import type { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer'; - -describe('PublishedWorkflowEnqueuer', () => { - const logger = mock(); - logger.scoped.mockReturnValue(logger); - - const outboxRepository = mock(); - const outboxConsumer = mock(); - - let enqueuer: PublishedWorkflowEnqueuer; - - function createEnqueuer(useWorkflowPublicationService = true) { - const workflowsConfig = mock({ useWorkflowPublicationService }); - return new PublishedWorkflowEnqueuer(logger, workflowsConfig, outboxRepository, outboxConsumer); - } - - beforeEach(() => { - vi.clearAllMocks(); - enqueuer = createEnqueuer(); - }); - - test('delegates to the outbox repository to enqueue active workflows requiring leader publication', async () => { - await enqueuer.enqueueActiveWorkflows(); - - expect(outboxRepository.enqueueForLeaderHandoff).toHaveBeenCalledTimes(1); - }); - - describe('enqueueOnLeaderTakeover', () => { - test('enqueues active workflows and drains immediately when the publication service is enabled', async () => { - await createEnqueuer(true).enqueueOnLeaderTakeover(); - - expect(outboxRepository.enqueueForLeaderHandoff).toHaveBeenCalledTimes(1); - expect(outboxConsumer.startPolling).toHaveBeenCalledTimes(1); - expect(outboxConsumer.drainPending).toHaveBeenCalledTimes(1); - }); - - test('does nothing when the publication service is disabled', async () => { - await createEnqueuer(false).enqueueOnLeaderTakeover(); - - expect(outboxRepository.enqueueForLeaderHandoff).not.toHaveBeenCalled(); - expect(outboxConsumer.startPolling).not.toHaveBeenCalled(); - expect(outboxConsumer.drainPending).not.toHaveBeenCalled(); - }); - }); -}); diff --git a/packages/cli/src/workflows/publication/__tests__/published-workflow-trigger-deactivator.test.ts b/packages/cli/src/workflows/publication/__tests__/published-workflow-trigger-deactivator.test.ts index 15cd9365e17..581a22bc370 100644 --- a/packages/cli/src/workflows/publication/__tests__/published-workflow-trigger-deactivator.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/published-workflow-trigger-deactivator.test.ts @@ -318,70 +318,5 @@ describe('PublishedWorkflowTriggerDeactivator', () => { expect(logger.warn).not.toHaveBeenCalled(); }); }); - - describe('interval lifecycle', () => { - beforeEach(() => { - vi.useFakeTimers(); - }); - - afterEach(() => { - vi.useRealTimers(); - }); - - const advanceOneInterval = async () => - await vi.advanceTimersByTimeAsync(RECONCILE_INTERVAL_SECONDS * 1000); - - test('sweeps every reconcile interval once started', async () => { - const deactivator = createDeactivator(); - deactivator.startGhostTriggerJanitor(); - - await advanceOneInterval(); - await advanceOneInterval(); - - expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).toHaveBeenCalledTimes(2); - - deactivator.stopGhostTriggerJanitor(); - }); - - test('does not start when the publication service is disabled', async () => { - createDeactivator(false).startGhostTriggerJanitor(); - - await advanceOneInterval(); - - expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled(); - }); - - test('starting twice schedules only one interval', async () => { - const deactivator = createDeactivator(); - deactivator.startGhostTriggerJanitor(); - deactivator.startGhostTriggerJanitor(); - - await advanceOneInterval(); - - expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).toHaveBeenCalledTimes(1); - - deactivator.stopGhostTriggerJanitor(); - }); - - test('stops sweeping once stopped', async () => { - const deactivator = createDeactivator(); - deactivator.startGhostTriggerJanitor(); - deactivator.stopGhostTriggerJanitor(); - - await advanceOneInterval(); - - expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled(); - }); - - test('does not restart after shutdown', async () => { - const deactivator = createDeactivator(); - deactivator.shutdown(); - deactivator.startGhostTriggerJanitor(); - - await advanceOneInterval(); - - expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled(); - }); - }); }); }); 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 0810a199dd8..2e423efc2b2 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 @@ -19,6 +19,7 @@ import type { import type { EventService } from '@/events/event.service'; import type { NonWebhookTriggerRegistrar } from '@/workflows/triggers/non-webhook-trigger-registrar'; +import type { PublishedWorkflowTriggerDeactivator } from '../published-workflow-trigger-deactivator'; import type { WorkflowPublicationLifecycleLock } from '../workflow-publication-lifecycle-lock'; import type { WorkflowPublicationOutboxConsumer } from '../workflow-publication-outbox-consumer'; import { WorkflowPublicationReconciler } from '../workflow-publication-reconciler.service'; @@ -39,6 +40,7 @@ const eventService = mock(); const lifecycleLock = mock(); const workflowRepository = mock(); const activeWorkflowTriggers = mock(); +const triggerDeactivator = mock(); let service: WorkflowPublicationReconciler; @@ -71,6 +73,7 @@ beforeEach(() => { workflowRepository.getActiveIds.mockResolvedValue([]); workflowRepository.findOneBy.mockResolvedValue(null); lifecycleLock.runExclusive.mockImplementation(async (_workflowId, fn) => await fn()); + triggerDeactivator.sweepGhostTriggers.mockResolvedValue(0); service = new WorkflowPublicationReconciler( logger, config, @@ -85,6 +88,7 @@ beforeEach(() => { lifecycleLock, workflowRepository, activeWorkflowTriggers, + triggerDeactivator, ); }); @@ -109,32 +113,89 @@ describe('WorkflowPublicationReconciler', () => { expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalledTimes(1); }); - it('does not start when the instance is not leader', () => { + it('schedules the loop on a follower too, ticking the ghost sweep instead of the detections', async () => { Object.assign(instanceSettings, { isLeader: false }); service.init(); - vi.advanceTimersByTime(10_000); + await vi.advanceTimersByTimeAsync(5_000); + expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalled(); expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); }); }); - describe('startReconciler', () => { - it('does not start when the publication service is disabled', () => { + describe('loop lifecycle', () => { + it('does not start when the publication service is disabled', async () => { Object.assign(config, { useWorkflowPublicationService: false }); + Object.assign(instanceSettings, { isLeader: false }); - service.startReconciler(); - vi.advanceTimersByTime(10_000); + service.init(); + await vi.advanceTimersByTimeAsync(10_000); + expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); }); - it('does not start if shutting down', () => { + it('does not start after shutdown', async () => { service.shutdown(); - service.startReconciler(); - vi.advanceTimersByTime(10_000); + service.init(); + await vi.advanceTimersByTimeAsync(10_000); + + expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); + expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); + }); + + it('keeps ticking across a stepdown, flipping to the follower checks', async () => { + service.init(); + await vi.advanceTimersByTimeAsync(5_000); + expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled(); + + Object.assign(instanceSettings, { isLeader: false }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalled(); + }); + + it('needs no takeover kick: the next tick after promotion runs the leader detections', async () => { + Object.assign(instanceSettings, { isLeader: false }); + service.init(); + + Object.assign(instanceSettings, { isLeader: true }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled(); + }); + + it('stops only at shutdown', async () => { + service.init(); + await vi.advanceTimersByTimeAsync(5_000); + + service.shutdown(); + triggerStatusRepository.findActivatedInMemoryTriggers.mockClear(); + await vi.advanceTimersByTimeAsync(20_000); expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); + expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); + }); + }); + + describe('leader takeover', () => { + it('kicks an immediate pass without waiting for the interval', async () => { + service.startReconciler(); + + await service.reconcileOnLeaderTakeover(); + + expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalledTimes(1); + }); + + it('does nothing while the loop is not running', async () => { + // The loop only runs when the publication service is enabled and the + // instance is not shutting down; the takeover kick must not open a + // side door around those gates. + await service.reconcileOnLeaderTakeover(); + + expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); + expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); }); }); @@ -233,12 +294,28 @@ describe('WorkflowPublicationReconciler', () => { ); }); - it('does not run when the instance is no longer leader', async () => { + it('runs the ghost sweep and none of the leader detections when not leader', async () => { Object.assign(instanceSettings, { isLeader: false }); await service.reconcile(); + expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalledTimes(1); expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); + expect(outboxRepository.findVersionSkewedWorkflowIds).not.toHaveBeenCalled(); + expect(outboxRepository.findTriggerStatusDriftedWorkflowIds).not.toHaveBeenCalled(); + expect(outboxRepository.findUnreportedPublishedWorkflowIds).not.toHaveBeenCalled(); + expect(outboxRepository.enqueueByWorkflowIds).not.toHaveBeenCalled(); + expect(eventService.emit).not.toHaveBeenCalledWith( + 'workflow-publication-reconciliation', + expect.anything(), + ); + }); + + it('does not run the ghost sweep on the leader', async () => { + await service.reconcile(); + + expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); + expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled(); }); }); @@ -328,15 +405,4 @@ describe('WorkflowPublicationReconciler', () => { expect(activeWorkflowTriggers.remove).not.toHaveBeenCalled(); }); }); - - describe('stopReconciler', () => { - it('stops the interval on leader stepdown', () => { - service.startReconciler(); - service.stopReconciler(); - - vi.advanceTimersByTime(10_000); - - expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); - }); - }); }); diff --git a/packages/cli/src/workflows/publication/published-workflow-enqueuer.ts b/packages/cli/src/workflows/publication/published-workflow-enqueuer.ts deleted file mode 100644 index 0f05aa95ad5..00000000000 --- a/packages/cli/src/workflows/publication/published-workflow-enqueuer.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { Logger } from '@n8n/backend-common'; -import { WorkflowsConfig } from '@n8n/config'; -import { WorkflowPublicationOutboxRepository } from '@n8n/db'; -import { OnLeaderTakeover } from '@n8n/decorators'; -import { Service } from '@n8n/di'; - -import { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer'; - -/** - * Enqueues a publication outbox record for every active workflow, on leader - * startup and on leader takeover (the two moments an instance becomes leader). - */ -@Service() -export class PublishedWorkflowEnqueuer { - constructor( - private readonly logger: Logger, - private readonly workflowsConfig: WorkflowsConfig, - private readonly outboxRepository: WorkflowPublicationOutboxRepository, - private readonly outboxConsumer: WorkflowPublicationOutboxConsumer, - ) { - this.logger = this.logger.scoped('workflow-publication'); - } - - /** - * Enqueue a publication outbox record for every active workflow that needs - * leader-side publication at its current active version. Workflows with only - * persisted (webhook) triggers are skipped: those live durably in - * `webhook_entity` and are served by any main, so a new leader has no - * in-memory state to rebuild for them. This is idempotent. - */ - async enqueueActiveWorkflows(): Promise { - await this.outboxRepository.enqueueForLeaderHandoff(); - this.logger.debug('Enqueued active workflow publication records'); - } - - /** - * On leader takeover, requeue every active workflow exactly like startup. Safe - * because publication is idempotent and a no-op publish emits no pubsub, so - * unchanged workflows stay quiet. - */ - @OnLeaderTakeover() - async enqueueOnLeaderTakeover(): Promise { - if (!this.workflowsConfig.useWorkflowPublicationService) return; - await this.enqueueActiveWorkflows(); - this.outboxConsumer.startPolling(); - await this.outboxConsumer.drainPending(); - } -} diff --git a/packages/cli/src/workflows/publication/published-workflow-trigger-deactivator.ts b/packages/cli/src/workflows/publication/published-workflow-trigger-deactivator.ts index 5a54fe2f857..1b535d0bf47 100644 --- a/packages/cli/src/workflows/publication/published-workflow-trigger-deactivator.ts +++ b/packages/cli/src/workflows/publication/published-workflow-trigger-deactivator.ts @@ -1,7 +1,6 @@ import { Logger } from '@n8n/backend-common'; import { WorkflowsConfig } from '@n8n/config'; -import { Time } from '@n8n/constants'; -import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; +import { OnLeaderStepdown, OnShutdown } from '@n8n/decorators'; import { Service } from '@n8n/di'; import { ActiveWorkflowTriggers, ErrorReporter, InstanceSettings } from 'n8n-core'; @@ -19,7 +18,8 @@ const STUCK_LOCK_WARN_AFTER_PASSES = 3; /** * Tears down in-memory triggers on leader stepdown and shutdown, and — as a - * safeguard — periodically sweeps the registry while not leader: a trigger + * safeguard — sweeps the registry while not leader (driven by the + * reconciler's loop, which ticks the sweep on non-leader instances): a trigger * registration on a non-leader should never exist, so anything found is torn * down and reported. * @@ -35,7 +35,6 @@ const STUCK_LOCK_WARN_AFTER_PASSES = 3; */ @Service() export class PublishedWorkflowTriggerDeactivator { - private ghostTriggerJanitorInterval: NodeJS.Timeout | undefined; private isShuttingDown = false; private consecutiveLockSkipsByWorkflowId: Map = new Map(); @@ -72,7 +71,7 @@ export class PublishedWorkflowTriggerDeactivator { const ghosts: string[] = []; - // Nothing may escape this method: the interval callback driving it has + // Nothing may escape this method: the loop tick driving it runs with // nobody awaiting it, so an escaped rejection would crash the process. try { const workflowIds = this.activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds(); @@ -132,30 +131,8 @@ export class PublishedWorkflowTriggerDeactivator { return deactivatedWorkflows; } - @OnLeaderStepdown() - startGhostTriggerJanitor(): void { - if (!this.workflowsConfig.useWorkflowPublicationService) return; - if (this.isShuttingDown) return; - if (this.ghostTriggerJanitorInterval) return; - this.consecutiveLockSkipsByWorkflowId = new Map(); - - this.ghostTriggerJanitorInterval = setInterval( - async () => await this.sweepGhostTriggers(), - this.workflowsConfig.publicationReconcileIntervalSeconds * Time.seconds.toMilliseconds, - ); - } - - @OnLeaderTakeover() - stopGhostTriggerJanitor(): void { - if (this.ghostTriggerJanitorInterval) { - clearInterval(this.ghostTriggerJanitorInterval); - this.ghostTriggerJanitorInterval = undefined; - } - } - @OnShutdown() shutdown(): void { this.isShuttingDown = true; - this.stopGhostTriggerJanitor(); } } 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 30e48388ffc..d2d2d860b84 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-outbox-consumer.ts @@ -1,7 +1,7 @@ import { Logger } from '@n8n/backend-common'; import { WorkflowsConfig } from '@n8n/config'; import { WorkflowPublicationOutbox, WorkflowPublicationOutboxRepository } from '@n8n/db'; -import { OnPubSubEvent, OnShutdown } from '@n8n/decorators'; +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'; @@ -93,6 +93,7 @@ export class WorkflowPublicationOutboxConsumer { * Also start polling - this is idempotent, so harmless if we're already polling. */ @OnPubSubEvent('workflow-publish-wake-up', { instanceType: 'main', instanceRole: 'leader' }) + @OnLeaderTakeover() async wakeUp(): Promise { if (!this.workflowsConfig.useWorkflowPublicationService) return; // Direct callers (e.g. a short-lived CLI process waking its own local consumer) diff --git a/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts b/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts index ac6dc0ebab5..11d7541afd0 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts @@ -6,7 +6,7 @@ import { WorkflowPublicationTriggerStatusRepository, WorkflowRepository, } from '@n8n/db'; -import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; +import { OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; import { Service } from '@n8n/di'; import { ActiveWorkflowTriggers, @@ -20,16 +20,23 @@ import type { WorkflowId } from 'n8n-workflow'; import { EventService } from '@/events/event.service'; import { NonWebhookTriggerRegistrar } from '@/workflows/triggers/non-webhook-trigger-registrar'; +import { PublishedWorkflowTriggerDeactivator } from './published-workflow-trigger-deactivator'; import { WorkflowPublicationLifecycleLock } from './workflow-publication-lifecycle-lock'; import { WorkflowPublicationOutboxConsumer } from './workflow-publication-outbox-consumer'; /** - * Periodically checks, on the leader, that every trigger which should be active - * in memory really is, and re-publishes any workflow whose triggers went - * missing. This recovers from the leader-transition edge case where a demoted - * instance consumed a workflow's outbox record (marking it completed) and then - * tore its triggers down, so the new leader never (re)activated them and no - * pending record remains to reprocess. + * The one reconciliation loop of the workflow publication service. It ticks on + * every main instance for the whole process lifetime; what a tick does is + * decided by the instance's role at that moment, so leadership transitions + * never start or stop anything — only shutdown stops the loop. + * + * Leader ticks verify that the running state matches the published state — + * every trigger that should be active in memory really is, and the publication + * tables agree with each other — and heal whatever diverged. Causes are + * deliberately not enumerated here: a known cause is a bug to fix at its + * source; this loop exists to detect and repair what an unknown one leaves + * behind. Non-leader ticks instead sweep the registry for ghost registrations + * (see {@link PublishedWorkflowTriggerDeactivator}). * * Detection is scoped to non-webhook (poll/trigger) nodes, since those are the * ones held in the in-memory registry; webhook triggers live in the @@ -37,16 +44,16 @@ import { WorkflowPublicationOutboxConsumer } from './workflow-publication-outbox * re-published. Recovery only enqueues through the outbox and wakes the consumer; * the applier does the actual re-registration, so this adds no activation logic. * - * A third detection compares versions rather than trigger node ids: a - * `workflow_published_version` mapping that disagrees with the workflow's - * `activeVersionId` (a rolled-back or lost publication, or a missed unpublish) - * is invisible to the node-id diffs when the trigger set is unchanged, so it is - * detected in the database directly and healed through the same enqueue path. - * Two further database detections complete the coverage: trigger-status rows - * recorded for a version other than the active one (the mapping may already be - * correct while the rows lag), and published workflows with no trigger-status - * rows at all (published before the service was enabled, or a publication that - * failed before reporting). + * The registry checks compare the trigger-status rows against what is + * registered in memory — and both were written by the last publication that + * was applied. They can tell when memory drifted from what was applied, but + * not whether what was applied is what the workflow's `activeVersionId` says + * it should be. Three database-only detections close that gap: a + * `workflow_published_version` mapping that disagrees with `activeVersionId`, + * trigger-status rows recorded for a version other than the active one, and + * published workflows with no trigger-status rows at all (also the expected + * state for anything published before the service was enabled). All heal + * through the same enqueue path. */ @Service() export class WorkflowPublicationReconciler { @@ -70,13 +77,12 @@ export class WorkflowPublicationReconciler { private readonly lifecycleLock: WorkflowPublicationLifecycleLock, private readonly workflowRepository: WorkflowRepository, private readonly activeWorkflowTriggers: ActiveWorkflowTriggers, + private readonly publishedWorkflowTriggerDeactivator: PublishedWorkflowTriggerDeactivator, ) { this.logger = logger.scoped('workflow-publication'); } init() { - if (!this.instanceSettings.isLeader) return; - this.startReconciler(); // Run an initial pass at startup rather than waiting a full interval, so a @@ -85,12 +91,10 @@ export class WorkflowPublicationReconciler { } /** - * Deliberately no immediate pass on takeover: `PublishedWorkflowEnqueuer` - * already re-enqueues every active workflow at that moment, and the - * consumed-record race this loop recovers can only materialize after that - * enqueue — the first interval pass catches it. + * Starts the process-lifetime tick interval, on any role. Takeover never + * starts or stops anything here — the loop is already ticking — but it does + * kick an immediate pass (see {@link reconcileOnLeaderTakeover}). */ - @OnLeaderTakeover() startReconciler() { if (!this.workflowsConfig.useWorkflowPublicationService) return; if (this.isShuttingDown || this.reconcileInterval) return; @@ -104,26 +108,28 @@ export class WorkflowPublicationReconciler { this.logger.debug(`Trigger reconciliation scheduled every ${intervalSeconds}s`); } - @OnLeaderStepdown() - stopReconciler() { + @OnShutdown() + shutdown() { + this.isShuttingDown = true; clearInterval(this.reconcileInterval); this.reconcileInterval = undefined; } - @OnShutdown() - shutdown() { - this.isShuttingDown = true; - this.stopReconciler(); - } - /** - * One reconciliation pass: diff the triggers that should be active in memory - * against what is registered, and re-publish any workflow with a missing - * trigger. Errors are caught and reported so a transient failure never escapes - * the interval; the next pass retries. Runs only on the leader. + * One tick of the loop, gated by the instance's role at this moment. On the + * leader: diff the triggers that should be active in memory against what is + * registered, and re-publish any workflow with a missing trigger. On any + * other role: sweep ghost registrations instead. Errors are caught and + * reported so a transient failure never escapes the interval; the next pass + * retries. */ async reconcile(): Promise { - if (!this.instanceSettings.isLeader || this.isShuttingDown) return; + if (this.isShuttingDown) return; + + if (!this.instanceSettings.isLeader) { + await this.publishedWorkflowTriggerDeactivator.sweepGhostTriggers(); + return; + } await this.tracing.startSpan( { name: 'Publication trigger reconciliation', op: 'publication.reconcile' }, @@ -185,6 +191,21 @@ export class WorkflowPublicationReconciler { ); } + /** + * A fresh leader must converge now, not an interval later: with an empty + * registry, this pass re-detects every workflow that needs its triggers + * back. Pending records left by the previous leader are excluded from the + * detections and drained by the consumer's own takeover hook instead. Gated + * on the loop running, so a disabled service or shutdown keeps takeover + * inert. + */ + @OnLeaderTakeover() + async reconcileOnLeaderTakeover(): Promise { + if (!this.reconcileInterval) return; + + await this.reconcile(); + } + private async findSurplusWorkflowIds(): Promise { const registered = this.activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds(); if (registered.length === 0) return []; diff --git a/packages/cli/test/integration/commands/start.cmd.test.ts b/packages/cli/test/integration/commands/start.cmd.test.ts index 9c2b80245c3..c4a904612b7 100644 --- a/packages/cli/test/integration/commands/start.cmd.test.ts +++ b/packages/cli/test/integration/commands/start.cmd.test.ts @@ -91,7 +91,7 @@ afterAll(async () => { ); Container.get(WorkflowPublicationOutboxConsumer).stopPolling(); Container.get(WorkflowPublicationOutboxCleanupService).stopCleanup(); - Container.get(WorkflowPublicationReconciler).stopReconciler(); + Container.get(WorkflowPublicationReconciler).shutdown(); await Container.get(ActiveWorkflowManager).removeAll(); await testDb.terminate(); }); diff --git a/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts b/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts index eea56f7af7e..7dda751047c 100644 --- a/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts @@ -1,23 +1,16 @@ -import { createActiveWorkflow, createWorkflow, testDb } from '@n8n/backend-test-utils'; +import { createActiveWorkflow, createWorkflow, newWorkflow, testDb } from '@n8n/backend-test-utils'; import { WorkflowsConfig } from '@n8n/config'; import { UNPUBLISH_VERSION_SENTINEL } from '@n8n/db'; -import type { WorkflowPublicationTriggerKind } from '@n8n/db'; -import { - WorkflowPublicationOutboxRepository, - WorkflowPublicationTriggerStatusRepository, - WorkflowRepository, -} from '@n8n/db'; +import { WorkflowPublicationOutboxRepository, WorkflowRepository } from '@n8n/db'; import { Container } from '@n8n/di'; import assert from 'node:assert'; describe('WorkflowPublicationOutboxRepository', () => { let repository: WorkflowPublicationOutboxRepository; - let triggerStatusRepository: WorkflowPublicationTriggerStatusRepository; beforeAll(async () => { await testDb.init(); repository = Container.get(WorkflowPublicationOutboxRepository); - triggerStatusRepository = Container.get(WorkflowPublicationTriggerStatusRepository); }); beforeEach(async () => { @@ -469,134 +462,24 @@ describe('WorkflowPublicationOutboxRepository', () => { expect(await repository.find({ where: { status: 'pending' } })).toHaveLength(0); }); - }); - describe('enqueueForLeaderHandoff', () => { - beforeEach(async () => { - await testDb.truncate([ - 'WorkflowPublicationTriggerStatus', - 'WorkflowDependency', - 'WorkflowEntity', - 'WorkflowHistory', - 'WorkflowPublishHistory', - ]); - }); + it('enqueues every workflow in a batch larger than one statement chunk', async () => { + // A fresh leader detects every active workflow as missing at once, so + // this call sees fleet-sized input. Sqlite binds one placeholder per id + // and caps bound variables per statement, so large batches are split + // into chunks; a boundary bug here would silently drop workflows. + const workflowIds = Array.from({ length: 1100 }, (_, i) => `bulk-wf-${i}`); + // Seed in slices: a single multi-row VALUES of this size exceeds sqlite's + // expression-tree depth, which is not the limit under test here. + for (let i = 0; i < workflowIds.length; i += 250) { + await Container.get(WorkflowRepository).insert( + workflowIds.slice(i, i + 250).map((id) => newWorkflow({ id })), + ); + } - // Seed a single trigger-status row of the given kind at the workflow's active - // version, appending to any already recorded for the workflow. - const recordTrigger = async ( - workflow: { id: string; activeVersionId: string | null }, - nodeId: string, - triggerKind: WorkflowPublicationTriggerKind, - status: 'activated' | 'failed' = 'activated', - ) => { - assert(workflow.activeVersionId); - const existing = await triggerStatusRepository.findByWorkflowId(workflow.id); - await triggerStatusRepository.replaceForWorkflow(workflow.id, [ - ...existing.map((row) => ({ - nodeId: row.nodeId, - versionId: row.versionId, - status: row.status, - triggerKind: row.triggerKind, - errorMessage: row.errorMessage, - })), - { - nodeId, - versionId: workflow.activeVersionId, - status, - triggerKind, - errorMessage: status === 'failed' ? 'boom' : null, - }, - ]); - }; + await repository.enqueueByWorkflowIds(workflowIds); - const pendingWorkflowIds = async () => { - const pending = await repository.find({ where: { status: 'pending' } }); - return pending.map((record) => record.workflowId); - }; - - it('enqueues a workflow whose recorded triggers include an in-memory trigger', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'in-memory'); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([workflow.id]); - }); - - it('skips a workflow whose recorded triggers are all persisted', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'persisted'); - await recordTrigger(workflow, 'n2', 'persisted'); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([]); - }); - - it('enqueues a workflow that has no recorded triggers yet', async () => { - const workflow = await createActiveWorkflow(); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([workflow.id]); - }); - - it('enqueues a workflow with a mix of persisted and in-memory triggers', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'persisted'); - await recordTrigger(workflow, 'n2', 'in-memory'); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([workflow.id]); - }); - - it('enqueues a workflow whose only in-memory trigger last failed to activate', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'in-memory', 'failed'); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([workflow.id]); - }); - - it('enqueues at the active version', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'in-memory'); - - await repository.enqueueForLeaderHandoff(); - - const pending = await repository.find({ where: { status: 'pending' } }); - expect(pending).toHaveLength(1); - expect(pending[0].publishedVersionId).toBe(workflow.activeVersionId); - }); - - it('skips inactive and archived workflows', async () => { - const active = await createActiveWorkflow(); - await recordTrigger(active, 'n1', 'in-memory'); - await createWorkflow(); // inactive: no activeVersionId - const archived = await createActiveWorkflow(); - await recordTrigger(archived, 'n1', 'in-memory'); - await Container.get(WorkflowRepository).update(archived.id, { isArchived: true }); - - await repository.enqueueForLeaderHandoff(); - - expect(await pendingWorkflowIds()).toEqual([active.id]); - }); - - it('is idempotent: re-running does not create duplicate pending records', async () => { - const workflow = await createActiveWorkflow(); - await recordTrigger(workflow, 'n1', 'in-memory'); - - await repository.enqueueForLeaderHandoff(); - await repository.enqueueForLeaderHandoff(); - - const pending = await repository.find({ - where: { workflowId: workflow.id, status: 'pending' }, - }); - expect(pending).toHaveLength(1); - expect(pending[0].publishedVersionId).toBe(workflow.activeVersionId); + expect(await repository.countBy({ status: 'pending' })).toBe(1100); }); }); diff --git a/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts b/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts index dbc141acd5c..8b844ac19ba 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-outbox-consumer.test.ts @@ -18,7 +18,6 @@ import { ExecutionService } from '@/executions/execution.service'; import { ExternalHooks } from '@/external-hooks'; import { Push } from '@/push'; import { OwnershipService } from '@/services/ownership.service'; -import { PublishedWorkflowEnqueuer } from '@/workflows/publication/published-workflow-enqueuer'; import { PublishedWorkflowTriggerDeactivator } from '@/workflows/publication/published-workflow-trigger-deactivator'; import { WorkflowPublicationLifecycleLock } from '@/workflows/publication/workflow-publication-lifecycle-lock'; import { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer'; @@ -207,28 +206,6 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); }); - test('startup enqueue + drain registers triggers for active workflows via reconciliation', async () => { - const owner = await createOwner(); - - const trigger = scheduleNode('startup'); - const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); - await setActiveVersion(workflow.id, workflow.versionId); - await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId); - - // Fresh leader startup: the workflow is active and published, but nothing is - // registered in memory yet. - expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined(); - - await Container.get(PublishedWorkflowEnqueuer).enqueueActiveWorkflows(); - consumer.startPolling(); - await consumer.drainPending(); - consumer.stopPolling(); - - // The reconciliation path registered the missing trigger and completed the record. - expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); - expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); - }); - test('does no trigger work when only non-trigger content changed', async () => { const owner = await createOwner(); diff --git a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts index 9f2c8c32784..0dea8c2743c 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts @@ -503,7 +503,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { ); }); - test('records persisted rows for a pseudo-only workflow, keeping it out of leader handoff and reconciliation', async () => { + test('records persisted rows for a pseudo-only workflow, keeping it out of reconciliation', async () => { const owner = await createOwner(); const trigger = manualTriggerNode('pseudo'); @@ -526,14 +526,9 @@ describe('WorkflowPublicationReconciler (integration)', () => { triggerKind: 'persisted', }); - // Leader handoff on a fresh leader: nothing is registered locally, but a - // pseudo-only workflow has no real trigger work to republish. + // Fresh leader: nothing is registered locally. The reconciler ignores + // persisted rows even though the node is not registered — no re-enqueue loop. await activeWorkflowTriggers.remove(workflow.id); - await outboxRepository.enqueueForLeaderHandoff(); - expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); - - // The reconciler ignores persisted rows even though the node is not - // registered — no re-enqueue loop. await reconciler.reconcile(); expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined();