feat(core): Detect and heal trigger-status drift and unreported published workflows in reconciliation (#35356)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts

@@ -266,6 +266,103 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
 		return rows.map((row) => row.workflowId);
 	}
 
+	/**
+	 * Published workflows where any trigger-status row was recorded for a version
+	 * other than the workflow's canonical `activeVersionId`. Catches a crash or a
+	 * stale (zombie) writer between the published-version advance and the status
+	 * report: the `workflow_published_version` mapping may already be correct, so
+	 * {@link findVersionSkewedWorkflowIds} cannot see it — only the rows lag.
+	 *
+	 * Workflows with an in-flight (pending/in_progress) record are excluded in
+	 * the same statement, for the same reason as the version-skew check: the
+	 * reporter only writes rows while its record is `in_progress`, so row drift
+	 * with no in-flight record is a real divergence, never a normal mid-flight
+	 * state.
+	 *
+	 * Workflows whose most recent record is terminal `failed` are excluded for
+	 * the same reason as in {@link findUnreportedPublishedWorkflowIds}: a
+	 * publication that deterministically fails before reporting leaves the rows
+	 * drifted forever, so re-enqueueing would loop every pass. A user republish
+	 * (a fresh pending record) still recovers.
+	 */
+	async findTriggerStatusDriftedWorkflowIds(): Promise<string[]> {
+		const outboxTableName = this.getTableName('workflow_publication_outbox');
+		const workflowTableName = this.getTableName('workflow_entity');
+		const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status');
+
+		// Both compared columns are non-null here (`activeVersionId` is filtered,
+		// `versionId` is NOT NULL), so plain `<>` needs no null-emulation.
+		const rows: Array<{ workflowId: string }> = await this.query(
+			`SELECT DISTINCT w."id" AS "workflowId"
+			 FROM ${workflowTableName} w
+			 INNER JOIN ${triggerStatusTableName} ts ON ts."workflowId" = w."id"
+			 WHERE w."activeVersionId" IS NOT NULL
+			 AND ts."versionId" <> w."activeVersionId"
+			 AND NOT EXISTS (
+				 SELECT 1 FROM ${outboxTableName} o
+				 WHERE o."workflowId" = w."id"
+				 AND o."status" IN ('${Status.Pending}', '${Status.InProgress}')
+			 )
+			 AND COALESCE((
+				 SELECT o."status" FROM ${outboxTableName} o
+				 WHERE o."workflowId" = w."id"
+				 ORDER BY o."id" DESC LIMIT 1
+			 ), '') <> '${Status.Failed}'`,
+		);
+
+		return rows.map((row) => row.workflowId);
+	}
+
+	/**
+	 * Published, non-archived workflows with no trigger-status rows at all — no
+	 * publication was ever reported for them. Two populations: workflows
+	 * published while the publication service flag was off (only the reporter
+	 * writes rows), and publications that terminally failed before reporting any
+	 * per-trigger status (a consumer-wrapped unexpected throw, `version-missing`).
+	 *
+	 * The failed population is why workflows whose most recent outbox record is
+	 * terminal `failed` are excluded: re-enqueueing them would fail before
+	 * reporting again, still leave zero rows, and so be re-detected on every
+	 * pass forever, reporting an error each round. Recovery is not lost — a user
+	 * republish enqueues a fresh pending record, which is excluded here as
+	 * in-flight while it processes normally. `partial_success` always writes
+	 * rows, so it never reaches this bucket.
+	 *
+	 * Same in-flight exclusion, in the same statement, as
+	 * {@link findVersionSkewedWorkflowIds}.
+	 */
+	async findUnreportedPublishedWorkflowIds(): Promise<string[]> {
+		const outboxTableName = this.getTableName('workflow_publication_outbox');
+		const workflowTableName = this.getTableName('workflow_entity');
+		const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status');
+
+		// "Most recent" is the highest id: ids are monotonically assigned, the
+		// same FIFO assumption the claim query relies on. Sqlite (>= 3.23) accepts
+		// the `false` literal, so one statement serves both dialects.
+		const rows: Array<{ workflowId: string }> = await this.query(
+			`SELECT w."id" AS "workflowId"
+			 FROM ${workflowTableName} w
+			 WHERE w."activeVersionId" IS NOT NULL
+			 AND w."isArchived" = false
+			 AND NOT EXISTS (
+				 SELECT 1 FROM ${triggerStatusTableName} ts
+				 WHERE ts."workflowId" = w."id"
+			 )
+			 AND NOT EXISTS (
+				 SELECT 1 FROM ${outboxTableName} o
+				 WHERE o."workflowId" = w."id"
+				 AND o."status" IN ('${Status.Pending}', '${Status.InProgress}')
+			 )
+			 AND COALESCE((
+				 SELECT o."status" FROM ${outboxTableName} o
+				 WHERE o."workflowId" = w."id"
+				 ORDER BY o."id" DESC LIMIT 1
+			 ), '') <> '${Status.Failed}'`,
+		);
+
+		return rows.map((row) => row.workflowId);
+	}
+
 	/**
 	 * Atomically claim the oldest pending record by transitioning its status to
 	 * `in_progress`. Postgres uses `FOR UPDATE SKIP LOCKED` so concurrent

packages/cli/src/events/maps/workflow-publication-metrics.event-map.ts

@@ -55,6 +55,10 @@ export type WorkflowPublicationMetricsEventMap = {
 		surplusCount: number;
 		/** Workflows re-enqueued because their published version diverged from the active version. */
 		versionSkewCount: number;
+		/** Workflows re-enqueued because a trigger-status row was recorded for a version other than the active one. */
+		statusDriftCount: number;
+		/** Published workflows re-enqueued because no publication ever reported trigger statuses for them. */
+		unreportedCount: number;
 		durationMs: number;
 	};
 

packages/cli/src/metrics/prometheus/workflow-publication-metrics.service.ts

@@ -210,6 +210,16 @@ export class PrometheusWorkflowPublicationMetricsService implements PrometheusMe
 			help: 'Total number of workflows re-enqueued by reconciliation because their published version diverged from the active version.',
 		});
 
+		const statusDrift = new promClient.Counter({
+			name: `${prefix}workflow_publication_reconciliation_status_drift_workflows_total`,
+			help: 'Total number of workflows re-enqueued by reconciliation because a trigger-status row was recorded for a version other than the active one.',
+		});
+
+		const unreported = new promClient.Counter({
+			name: `${prefix}workflow_publication_reconciliation_unreported_workflows_total`,
+			help: 'Total number of published workflows re-enqueued by reconciliation because no publication had reported trigger statuses for them.',
+		});
+
 		const duration = new promClient.Histogram({
 			name: `${prefix}workflow_publication_reconciliation_duration_seconds`,
 			help: 'Duration in seconds of a trigger reconciliation pass by result.',
@@ -219,10 +229,20 @@ export class PrometheusWorkflowPublicationMetricsService implements PrometheusMe
 
 		this.eventService.on(
 			'workflow-publication-reconciliation',
-			({ result, deficientCount, surplusCount, versionSkewCount, durationMs }) => {
+			({
+				result,
+				deficientCount,
+				surplusCount,
+				versionSkewCount,
+				statusDriftCount,
+				unreportedCount,
+				durationMs,
+			}) => {
 				deficient.inc(deficientCount);
 				surplus.inc(surplusCount);
 				versionSkew.inc(versionSkewCount);
+				statusDrift.inc(statusDriftCount);
+				unreported.inc(unreportedCount);
 				duration.observe({ result }, durationMs * Time.milliseconds.toSeconds);
 			},
 		);

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

@@ -62,6 +62,8 @@ beforeEach(() => {
 	outboxRepository.enqueueByWorkflowIds.mockResolvedValue();
 	outboxRepository.findInFlightByWorkflowId.mockResolvedValue(null);
 	outboxRepository.findVersionSkewedWorkflowIds.mockResolvedValue([]);
+	outboxRepository.findTriggerStatusDriftedWorkflowIds.mockResolvedValue([]);
+	outboxRepository.findUnreportedPublishedWorkflowIds.mockResolvedValue([]);
 	outboxConsumer.drainPending.mockResolvedValue(0);
 	setRegistered({});
 	activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds.mockReturnValue([]);
@@ -192,6 +194,32 @@ describe('WorkflowPublicationReconciler', () => {
 			);
 		});
 
+		it('re-enqueues a workflow whose trigger-status rows lag the active version', async () => {
+			outboxRepository.findTriggerStatusDriftedWorkflowIds.mockResolvedValue(['wf-drift']);
+
+			await service.reconcile();
+
+			expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-drift']);
+			expect(outboxConsumer.drainPending).toHaveBeenCalled();
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'workflow-publication-reconciliation',
+				expect.objectContaining({ result: 'success', statusDriftCount: 1, versionSkewCount: 0 }),
+			);
+		});
+
+		it('re-enqueues a published workflow that has no trigger-status rows', async () => {
+			outboxRepository.findUnreportedPublishedWorkflowIds.mockResolvedValue(['wf-unreported']);
+
+			await service.reconcile();
+
+			expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-unreported']);
+			expect(outboxConsumer.drainPending).toHaveBeenCalled();
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'workflow-publication-reconciliation',
+				expect.objectContaining({ result: 'success', unreportedCount: 1, statusDriftCount: 0 }),
+			);
+		});
+
 		it('catches and reports errors without throwing', async () => {
 			triggerStatusRepository.findActivatedInMemoryTriggers.mockRejectedValue(
 				new Error('DB error'),

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

@@ -42,6 +42,11 @@ import { WorkflowPublicationOutboxConsumer } from './workflow-publication-outbox
  * `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).
  */
 @Service()
 export class WorkflowPublicationReconciler {
@@ -136,10 +141,22 @@ export class WorkflowPublicationReconciler {
 						await this.outboxRepository.findVersionSkewedWorkflowIds(),
 						'Re-enqueuing workflows whose published version diverged from the active version',
 					);
+					// Also after the earlier drains, for the same reason: rows those
+					// passes already converged must not be re-detected here.
+					const statusDrift = await this.republishWorkflows(
+						await this.outboxRepository.findTriggerStatusDriftedWorkflowIds(),
+						'Re-enqueuing workflows whose trigger-status rows lag the active version',
+					);
+					const unreported = await this.republishWorkflows(
+						await this.outboxRepository.findUnreportedPublishedWorkflowIds(),
+						'Re-enqueuing published workflows that no publication reported trigger statuses for',
+					);
 
 					span.setAttribute('n8n.publication.deficient_workflows', missing);
 					span.setAttribute('n8n.publication.surplus_workflows', surplus);
 					span.setAttribute('n8n.publication.version_skewed_workflows', versionSkew);
+					span.setAttribute('n8n.publication.status_drifted_workflows', statusDrift);
+					span.setAttribute('n8n.publication.unreported_workflows', unreported);
 
 					span.setStatus({ code: SpanStatus.ok });
 					this.eventService.emit('workflow-publication-reconciliation', {
@@ -147,6 +164,8 @@ export class WorkflowPublicationReconciler {
 						deficientCount: missing,
 						surplusCount: surplus,
 						versionSkewCount: versionSkew,
+						statusDriftCount: statusDrift,
+						unreportedCount: unreported,
 						durationMs: Date.now() - startedAt,
 					});
 				} catch (error) {
@@ -157,6 +176,8 @@ export class WorkflowPublicationReconciler {
 						deficientCount: 0,
 						surplusCount: 0,
 						versionSkewCount: 0,
+						statusDriftCount: 0,
+						unreportedCount: 0,
 						durationMs: Date.now() - startedAt,
 					});
 				}

packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts

@@ -351,6 +351,158 @@ describe('WorkflowPublicationReconciler (integration)', () => {
 		expect(await outboxRepository.findVersionSkewedWorkflowIds()).not.toContain(workflow.id);
 	});
 
+	test('re-writes trigger-status rows recorded for a version other than the active one', async () => {
+		const owner = await createOwner();
+
+		const trigger = scheduleNode('status-drift');
+		const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
+		await setActiveVersion(workflow.id, workflow.versionId);
+
+		await outboxRepository.enqueue(workflow.id, workflow.versionId);
+		await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!);
+
+		const newVersionId = 'version-2-status-drift';
+		await createWorkflowHistory(workflow, owner, undefined, {
+			versionId: newVersionId,
+			nodes: [{ ...trigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }],
+		});
+		await setActiveVersion(workflow.id, newVersionId);
+		await outboxRepository.enqueue(workflow.id, newVersionId);
+		await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!);
+
+		// A zombie writer rewrites the status rows for the old version after its
+		// record already resolved. The mapping still agrees with `activeVersionId`,
+		// so the version-skew check cannot see this — only the rows lag.
+		await triggerStatusRepository.update(
+			{ workflowId: workflow.id },
+			{ versionId: workflow.versionId },
+		);
+		expect(await outboxRepository.findVersionSkewedWorkflowIds()).not.toContain(workflow.id);
+
+		await reconciler.reconcile();
+
+		const rows = await triggerStatusRepository.findByWorkflowId(workflow.id);
+		expect(rows).toHaveLength(1);
+		expect(rows[0].versionId).toBe(newVersionId);
+		expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
+	});
+
+	test('publishes a workflow that was published while the publication service was off', async () => {
+		const owner = await createOwner();
+
+		const trigger = scheduleNode('pre-flag');
+		const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
+		await setActiveVersion(workflow.id, workflow.versionId);
+		// Legacy activation maintains the published-version mapping, but only the
+		// publication reporter writes trigger-status rows — so a workflow published
+		// while the flag was off has none, and no outbox record either.
+		await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId);
+
+		await reconciler.reconcile();
+
+		expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true);
+		const rows = await triggerStatusRepository.findByWorkflowId(workflow.id);
+		expect(rows).toHaveLength(1);
+		expect(rows[0]).toMatchObject({ nodeId: trigger.id, status: 'activated' });
+		expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
+	});
+
+	test('leaves a workflow whose most recent publication failed before reporting statuses alone', async () => {
+		const owner = await createOwner();
+
+		const trigger = scheduleNode('failed-terminal');
+		const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
+		await setActiveVersion(workflow.id, workflow.versionId);
+		// A publish that crashed after advancing the mapping but before the
+		// reporter wrote any rows: mapping equals `activeVersionId`, the record is
+		// terminal `failed`, zero rows. Re-enqueueing would fail before reporting
+		// again and still leave zero rows — an every-pass loop — so the pass must
+		// leave it for a user republish (a fresh pending record) to recover.
+		await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId);
+		await outboxRepository.enqueue(workflow.id, workflow.versionId);
+		const record = await outboxRepository.claimNextPendingRecord();
+		await outboxRepository.markFailed(record!.id, 'unexpected error before reporting');
+
+		await reconciler.reconcile();
+
+		expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
+		expect(await triggerStatusRepository.findByWorkflowId(workflow.id)).toHaveLength(0);
+	});
+
+	test('leaves a drifted workflow whose most recent publication failed before reporting alone', async () => {
+		const owner = await createOwner();
+
+		const trigger = scheduleNode('drift-failed-terminal');
+		const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
+		await setActiveVersion(workflow.id, workflow.versionId);
+
+		await outboxRepository.enqueue(workflow.id, workflow.versionId);
+		await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!);
+
+		// A v2 publish that crashed after advancing the mapping but before the
+		// reporter rewrote the rows: mapping equals `activeVersionId`, the record
+		// is terminal `failed`, and the rows still carry v1. If the failure is
+		// deterministic, re-enqueueing loops — the pass must leave it for a user
+		// republish to recover, exactly like the zero-rows case.
+		const newVersionId = 'version-2-drift-failed';
+		await createWorkflowHistory(workflow, owner, undefined, {
+			versionId: newVersionId,
+			nodes: [{ ...trigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }],
+		});
+		await setActiveVersion(workflow.id, newVersionId);
+		await publishedVersionRepository.setPublishedVersion(workflow.id, newVersionId);
+		await outboxRepository.enqueue(workflow.id, newVersionId);
+		const record = await outboxRepository.claimNextPendingRecord();
+		await outboxRepository.markFailed(record!.id, 'unexpected error before reporting');
+
+		await reconciler.reconcile();
+
+		expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
+		const rows = await triggerStatusRepository.findByWorkflowId(workflow.id);
+		expect(rows[0].versionId).toBe(workflow.versionId);
+	});
+
+	test('leaves drifted and unreported workflows with an in-flight record for that record to converge', async () => {
+		const owner = await createOwner();
+
+		// Drifted rows, but a pending publish owns the convergence.
+		const driftTrigger = scheduleNode('drift-in-flight');
+		const drifted = await createWorkflowWithHistory({ active: true, nodes: [driftTrigger] }, owner);
+		await setActiveVersion(drifted.id, drifted.versionId);
+		await outboxRepository.enqueue(drifted.id, drifted.versionId);
+		await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!);
+		const newVersionId = 'version-2-drift-in-flight';
+		await createWorkflowHistory(drifted, owner, undefined, {
+			versionId: newVersionId,
+			nodes: [{ ...driftTrigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }],
+		});
+		await setActiveVersion(drifted.id, newVersionId);
+		await outboxRepository.enqueue(drifted.id, newVersionId);
+
+		// No rows yet, but the first publish is still pending.
+		const unreportedTrigger = scheduleNode('unreported-in-flight');
+		const unreported = await createWorkflowWithHistory(
+			{ active: true, nodes: [unreportedTrigger] },
+			owner,
+		);
+		await setActiveVersion(unreported.id, unreported.versionId);
+		await outboxRepository.enqueue(unreported.id, unreported.versionId);
+
+		expect(await outboxRepository.findTriggerStatusDriftedWorkflowIds()).not.toContain(drifted.id);
+		expect(await outboxRepository.findUnreportedPublishedWorkflowIds()).not.toContain(
+			unreported.id,
+		);
+
+		await reconciler.reconcile();
+
+		// Untouched: both records are still pending — reconciliation neither
+		// enqueued nor drained.
+		expect((await outboxRepository.findInFlightByWorkflowId(drifted.id))?.status).toBe('pending');
+		expect((await outboxRepository.findInFlightByWorkflowId(unreported.id))?.status).toBe(
+			'pending',
+		);
+	});
+
 	test('records persisted rows for a pseudo-only workflow, keeping it out of leader handoff and reconciliation', async () => {
 		const owner = await createOwner();