From fd5ecfff6d697d2b8e56db4aa99256e4077e8c5c Mon Sep 17 00:00:00 2001 From: Danny Martini Date: Mon, 3 Aug 2026 16:24:51 +0200 Subject: [PATCH] feat(core): Detect and heal trigger-status drift and unreported published workflows in reconciliation (#35356) Co-authored-by: Claude Fable 5 --- .../workflow-publication-outbox.repository.ts | 97 +++++++++++ .../workflow-publication-metrics.event-map.ts | 4 + .../workflow-publication-metrics.service.ts | 22 ++- ...low-publication-reconciler.service.test.ts | 28 ++++ ...workflow-publication-reconciler.service.ts | 21 +++ .../workflow-publication-reconciler.test.ts | 152 ++++++++++++++++++ 6 files changed, 323 insertions(+), 1 deletion(-) 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 059de4eca22..8fe3b8b1b30 100644 --- a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts @@ -266,6 +266,103 @@ export class WorkflowPublicationOutboxRepository extends Repository 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 { + 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 { + 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 diff --git a/packages/cli/src/events/maps/workflow-publication-metrics.event-map.ts b/packages/cli/src/events/maps/workflow-publication-metrics.event-map.ts index 340270b2b90..a648a111d42 100644 --- a/packages/cli/src/events/maps/workflow-publication-metrics.event-map.ts +++ b/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; }; diff --git a/packages/cli/src/metrics/prometheus/workflow-publication-metrics.service.ts b/packages/cli/src/metrics/prometheus/workflow-publication-metrics.service.ts index 4e92e1687da..9adf9eec5a7 100644 --- a/packages/cli/src/metrics/prometheus/workflow-publication-metrics.service.ts +++ b/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); }, ); 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 accba96642b..0810a199dd8 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 @@ -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'), 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 0fded3e0f8d..ac6dc0ebab5 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts +++ b/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, }); } 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 9730c6a4ae7..9f2c8c32784 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts +++ b/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();