diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 5761ad1d3a3..74fc1d18c18 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -123,7 +123,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.workflow_dependency](public.workflow_dependency.md) | 9 | | BASE TABLE | | [public.workflow_entity](public.workflow_entity.md) | 20 | | BASE TABLE | | [public.workflow_history](public.workflow_history.md) | 11 | | BASE TABLE | -| [public.workflow_publication_outbox](public.workflow_publication_outbox.md) | 7 | | BASE TABLE | +| [public.workflow_publication_outbox](public.workflow_publication_outbox.md) | 8 | | BASE TABLE | | [public.workflow_publication_trigger_status](public.workflow_publication_trigger_status.md) | 8 | | BASE TABLE | | [public.workflow_publish_history](public.workflow_publish_history.md) | 6 | | BASE TABLE | | [public.workflow_published_version](public.workflow_published_version.md) | 4 | | BASE TABLE | @@ -1497,6 +1497,7 @@ erDiagram text errorMessage integer id varchar_36_ publishedVersionId + varchar_20_ reason varchar_20_ status timestamp_3__with_time_zone updatedAt varchar_36_ workflowId diff --git a/docs/generated/postgres-schema/public.workflow_publication_outbox.md b/docs/generated/postgres-schema/public.workflow_publication_outbox.md index ad7fa1c44f9..7a1ee69bac5 100644 --- a/docs/generated/postgres-schema/public.workflow_publication_outbox.md +++ b/docs/generated/postgres-schema/public.workflow_publication_outbox.md @@ -8,6 +8,7 @@ | errorMessage | text | | true | | | Error details for surfacing failed publications to the user. | | id | integer | | false | | | | | publishedVersionId | varchar(36) | | false | | | References workflow_history.versionId. | +| reason | varchar(20) | 'publish'::character varying | false | | | Why this publication was enqueued: user publish/unpublish, leader startup pass, leadership takeover, or reconciler drift-heal | | status | varchar(20) | | false | | | | | updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | | workflowId | varchar(36) | | false | | | References workflow_entity.id. | @@ -16,11 +17,13 @@ | Name | Type | Definition | | ---- | ---- | ---------- | +| CHK_workflow_publication_outbox_reason | CHECK | CHECK (((reason)::text = ANY ((ARRAY['publish'::character varying, 'startup'::character varying, 'leadership-takeover'::character varying, 'reconcile'::character varying])::text[]))) | | CHK_workflow_publication_outbox_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['pending'::character varying, 'in_progress'::character varying, 'completed'::character varying, 'partial_success'::character varying, 'failed'::character varying])::text[]))) | | PK_b3e2eeee36a4bd044d56468d311 | PRIMARY KEY | PRIMARY KEY (id) | | workflow_publication_outbox_createdAt_not_null | n | NOT NULL "createdAt" | | workflow_publication_outbox_id_not_null | n | NOT NULL id | | workflow_publication_outbox_publishedVersionId_not_null | n | NOT NULL "publishedVersionId" | +| workflow_publication_outbox_reason_not_null | n | NOT NULL reason | | workflow_publication_outbox_status_not_null | n | NOT NULL status | | workflow_publication_outbox_updatedAt_not_null | n | NOT NULL "updatedAt" | | workflow_publication_outbox_workflowId_not_null | n | NOT NULL "workflowId" | @@ -43,6 +46,7 @@ erDiagram text errorMessage integer id varchar_36_ publishedVersionId + varchar_20_ reason varchar_20_ status timestamp_3__with_time_zone updatedAt varchar_36_ workflowId diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index ac9893f8880..a99739a6d8c 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -123,7 +123,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [workflow_dependency](workflow_dependency.md) | 9 | | table | | [workflow_entity](workflow_entity.md) | 20 | | table | | [workflow_history](workflow_history.md) | 11 | | table | -| [workflow_publication_outbox](workflow_publication_outbox.md) | 7 | | table | +| [workflow_publication_outbox](workflow_publication_outbox.md) | 8 | | table | | [workflow_publication_trigger_status](workflow_publication_trigger_status.md) | 8 | | table | | [workflow_publish_history](workflow_publish_history.md) | 6 | | table | | [workflow_published_version](workflow_published_version.md) | 4 | | table | @@ -1486,6 +1486,7 @@ erDiagram TEXT errorMessage INTEGER id varchar_36_ publishedVersionId + varchar_20_ reason varchar_20_ status datetime_3_ updatedAt varchar_36_ workflowId diff --git a/docs/generated/sqlite-schema/workflow_publication_outbox.md b/docs/generated/sqlite-schema/workflow_publication_outbox.md index cabe5eb35de..ceea73d46a0 100644 --- a/docs/generated/sqlite-schema/workflow_publication_outbox.md +++ b/docs/generated/sqlite-schema/workflow_publication_outbox.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "workflow_publication_outbox" ("id" integer PRIMARY KEY NOT NULL, "workflowId" varchar(36) NOT NULL, "publishedVersionId" varchar(36) NOT NULL, "status" varchar(20) NOT NULL, "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_workflow_publication_outbox_status" CHECK ("status" IN ('pending', 'in_progress', 'completed', 'partial_success', 'failed'))) +CREATE TABLE "workflow_publication_outbox" ("id" integer PRIMARY KEY NOT NULL, "workflowId" varchar(36) NOT NULL, "publishedVersionId" varchar(36) NOT NULL, "status" varchar(20) NOT NULL, "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "reason" varchar(20) NOT NULL DEFAULT ('publish'), CONSTRAINT "CHK_workflow_publication_outbox_status" CHECK (("status" IN ('pending', 'in_progress', 'completed', 'partial_success', 'failed'))), CONSTRAINT "CHK_workflow_publication_outbox_reason" CHECK ("reason" IN ('publish', 'startup', 'leadership-takeover', 'reconcile'))) ``` @@ -19,6 +19,7 @@ CREATE TABLE "workflow_publication_outbox" ("id" integer PRIMARY KEY NOT NULL, " | errorMessage | TEXT | | true | | | | | id | INTEGER | | false | | | | | publishedVersionId | varchar(36) | | false | | | | +| reason | varchar(20) | 'publish' | false | | | | | status | varchar(20) | | false | | | | | updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | | workflowId | varchar(36) | | false | | | | @@ -27,7 +28,8 @@ CREATE TABLE "workflow_publication_outbox" ("id" integer PRIMARY KEY NOT NULL, " | Name | Type | Definition | | ---- | ---- | ---------- | -| - | CHECK | CHECK ("status" IN ('pending', 'in_progress', 'completed', 'partial_success', 'failed')) | +| - | CHECK | CHECK (("status" IN ('pending', 'in_progress', 'completed', 'partial_success', 'failed'))) | +| - | CHECK | CHECK ("reason" IN ('publish', 'startup', 'leadership-takeover', 'reconcile')) | | id | PRIMARY KEY | PRIMARY KEY (id) | ## Indexes @@ -47,6 +49,7 @@ erDiagram TEXT errorMessage INTEGER id varchar_36_ publishedVersionId + varchar_20_ reason varchar_20_ status datetime_3_ updatedAt varchar_36_ workflowId diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index d563cd12574..d5ecbd03d8c 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/packages/@n8n/db/src/entities/index.ts @@ -65,6 +65,7 @@ import { UNPUBLISH_VERSION_SENTINEL, WorkflowPublicationOutbox, WorkflowPublicationOutboxStatus, + WorkflowPublicationReason, } from './workflow-publication-outbox'; import { WorkflowPublicationTriggerStatus, @@ -146,6 +147,7 @@ export { WorkflowHistory, WorkflowPublicationOutbox, WorkflowPublicationOutboxStatus, + WorkflowPublicationReason, UNPUBLISH_VERSION_SENTINEL, WorkflowPublicationTriggerStatus, type WorkflowPublicationTriggerStatusType, diff --git a/packages/@n8n/db/src/entities/workflow-publication-outbox.ts b/packages/@n8n/db/src/entities/workflow-publication-outbox.ts index 66fb1740178..77a3c238bc7 100644 --- a/packages/@n8n/db/src/entities/workflow-publication-outbox.ts +++ b/packages/@n8n/db/src/entities/workflow-publication-outbox.ts @@ -21,6 +21,25 @@ export const WorkflowPublicationOutboxStatus = { export type WorkflowPublicationOutboxStatus = (typeof WorkflowPublicationOutboxStatus)[keyof typeof WorkflowPublicationOutboxStatus]; +/** + * Why a publication record was enqueued. The applier translates it into the + * `WorkflowActivateMode` reported to trigger nodes, so e.g. the n8n Trigger's + * "Instance Started" event fires only for the leader's startup pass. + */ +export const WorkflowPublicationReason = { + /** A user published or unpublished the workflow. */ + Publish: 'publish', + /** The leader's initial pass after the process booted. */ + Startup: 'startup', + /** A running instance became leader without restarting. */ + LeadershipTakeover: 'leadership-takeover', + /** A periodic reconciliation pass healing detected drift. */ + Reconcile: 'reconcile', +} as const; + +export type WorkflowPublicationReason = + (typeof WorkflowPublicationReason)[keyof typeof WorkflowPublicationReason]; + @Entity({ name: 'workflow_publication_outbox' }) @Index('IDX_workflow_publication_outbox_active_workflow_status', ['workflowId', 'status'], { unique: true, @@ -39,6 +58,9 @@ export class WorkflowPublicationOutbox extends WithTimestamps { @Column({ type: 'varchar', length: 20 }) status: WorkflowPublicationOutboxStatus; + @Column({ type: 'varchar', length: 20, default: 'publish' }) + reason: WorkflowPublicationReason; + @Column({ type: 'text', nullable: true }) errorMessage: string | null; } diff --git a/packages/@n8n/db/src/migrations/common/1786519946974-AddReasonToWorkflowPublicationOutbox.ts b/packages/@n8n/db/src/migrations/common/1786519946974-AddReasonToWorkflowPublicationOutbox.ts new file mode 100644 index 00000000000..d6e1566bf8c --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1786519946974-AddReasonToWorkflowPublicationOutbox.ts @@ -0,0 +1,37 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const tableName = 'workflow_publication_outbox'; +const columnName = 'reason'; + +/** + * Adds `reason` so each publication record carries why it was enqueued: a user + * publish/unpublish, the leader's startup pass, a leadership takeover, or the + * reconciler healing drift. The applier translates it into the activation mode + * reported to trigger nodes (e.g. the n8n Trigger's "Instance Started" event + * fires only for `startup`). + * + * NOT NULL with a `publish` default so rows inserted by pre-upgrade instances + * during a rolling deploy keep today's behavior (activation mode `update`). + */ +export class AddReasonToWorkflowPublicationOutbox1786519946974 implements ReversibleMigration { + async up({ schemaBuilder: { addColumns, column } }: MigrationContext) { + await addColumns( + tableName, + [ + column(columnName) + .varchar(20) + .notNull.default("'publish'") + .withEnumCheck(['publish', 'startup', 'leadership-takeover', 'reconcile']) + .comment( + 'Why this publication was enqueued: user publish/unpublish, leader startup pass, leadership takeover, or reconciler drift-heal', + ), + ], + { recreatesOnSqlite: true }, + ); + } + + async down({ schemaBuilder: { dropColumns, dropEnumCheck } }: MigrationContext) { + await dropEnumCheck(tableName, columnName, { recreatesOnSqlite: true }); + await dropColumns(tableName, [columnName], { recreatesOnSqlite: true }); + } +} 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 c74e7f808aa..43f261dc424 100644 --- a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts @@ -8,11 +8,12 @@ import { UNPUBLISH_VERSION_SENTINEL, WorkflowPublicationOutbox, WorkflowPublicationOutboxStatus as Status, + type WorkflowPublicationReason, } 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; +/** Sqlite bound-variable budget per statement (ids + the reason); safely under every build's cap. */ +const SQLITE_ENQUEUE_CHUNK_SIZE = 998; @Service() export class WorkflowPublicationOutboxRepository extends Repository { @@ -47,23 +48,33 @@ export class WorkflowPublicationOutboxRepository extends Repository { if (this.globalConfig.database.type === 'postgresdb') { - await this.enqueueWithPostgresUpsert(workflowId, publishedVersionId, trx ?? this.manager); + await this.enqueueWithPostgresUpsert( + workflowId, + publishedVersionId, + reason, + trx ?? this.manager, + ); return; } - await this.enqueueWithSqliteUpsert(workflowId, publishedVersionId, trx ?? this.manager); + await this.enqueueWithSqliteUpsert(workflowId, publishedVersionId, reason, trx ?? this.manager); } private async enqueueWithPostgresUpsert( workflowId: string, publishedVersionId: string, + reason: WorkflowPublicationReason, trx: EntityManager, ): Promise { const tableName = this.getTableName('workflow_publication_outbox'); @@ -71,27 +82,28 @@ export class WorkflowPublicationOutboxRepository extends Repository { const tableName = this.getTableName('workflow_publication_outbox'); await trx.query( - `INSERT INTO ${tableName} ("workflowId", "publishedVersionId", "status") - VALUES (?, ?, '${Status.Pending}') + `INSERT INTO ${tableName} ("workflowId", "publishedVersionId", "status", "reason") + VALUES (?, ?, '${Status.Pending}', ?) 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')`, - [workflowId, publishedVersionId], + DO UPDATE SET "publishedVersionId" = excluded."publishedVersionId", "reason" = excluded."reason", "updatedAt" = STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')`, + [workflowId, publishedVersionId, reason], ); } @@ -104,18 +116,24 @@ export class WorkflowPublicationOutboxRepository extends Repository { + async enqueueByWorkflowIds( + workflowIds: string[], + reason: WorkflowPublicationReason, + ): Promise { if (workflowIds.length === 0) return; if (this.globalConfig.database.type === 'postgresdb') { - await this.enqueueByWorkflowIdsWithPostgresUpsert(workflowIds); + await this.enqueueByWorkflowIdsWithPostgresUpsert(workflowIds, reason); return; } - await this.enqueueByWorkflowIdsWithSqliteUpsert(workflowIds); + await this.enqueueByWorkflowIdsWithSqliteUpsert(workflowIds, reason); } - private async enqueueByWorkflowIdsWithPostgresUpsert(workflowIds: string[]): Promise { + private async enqueueByWorkflowIdsWithPostgresUpsert( + workflowIds: string[], + reason: WorkflowPublicationReason, + ): Promise { const outboxTableName = this.getTableName('workflow_publication_outbox'); const workflowTableName = this.getTableName('workflow_entity'); @@ -131,17 +149,20 @@ export class WorkflowPublicationOutboxRepository extends Repository { + private async enqueueByWorkflowIdsWithSqliteUpsert( + workflowIds: string[], + reason: WorkflowPublicationReason, + ): Promise { const outboxTableName = this.getTableName('workflow_publication_outbox'); const workflowTableName = this.getTableName('workflow_entity'); @@ -154,13 +175,13 @@ export class WorkflowPublicationOutboxRepository extends Repository '?').join(', '); await this.query( - `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status") - SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}' + `INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status", "reason") + 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, + [reason, ...chunk], ); } } diff --git a/packages/cli/src/workflows/__tests__/workflow.service.test.ts b/packages/cli/src/workflows/__tests__/workflow.service.test.ts index 45f8601d24e..84bc9257738 100644 --- a/packages/cli/src/workflows/__tests__/workflow.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow.service.test.ts @@ -1373,6 +1373,7 @@ describe('WorkflowService', () => { expect(outboxRepositoryMock.enqueue).toHaveBeenCalledWith( WORKFLOW_ID, TARGET_VERSION_ID, + 'publish', trx, ); // publish-history records (deactivated for the previous version, activated for the diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts index 0c4f443aacd..6f5a80e3655 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts @@ -283,6 +283,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['b']), + 'update', ); expect(workflowPublishedVersionRepository.setPublishedVersion).toHaveBeenCalledWith( 'wf-1', @@ -290,6 +291,39 @@ describe('WorkflowPublicationApplier', () => { ); }); + describe('activation mode from the record reason', () => { + test.each([ + ['publish', 'update'], + ['startup', 'init'], + ['leadership-takeover', 'leadershipChange'], + ['reconcile', 'update'], + ] as const)('reason %s activates with mode %s', async (reason, expectedMode) => { + setTriggerSets([], [triggerNode('a')]); + + await applier.apply(makeRecord({ reason })); + + expect(workflowTriggerActivator.activate).toHaveBeenCalledWith( + expect.objectContaining({ id: 'wf-1' }), + newVersion, + new Set(['a']), + expectedMode, + ); + }); + + test('a record without a reason (pre-migration row) activates with update', async () => { + setTriggerSets([], [triggerNode('a')]); + + await applier.apply(makeRecord({ reason: undefined })); + + expect(workflowTriggerActivator.activate).toHaveBeenCalledWith( + expect.objectContaining({ id: 'wf-1' }), + newVersion, + new Set(['a']), + 'update', + ); + }); + }); + test('reconciles by registering desired non-webhook triggers missing from memory', async () => { // Same version on both sides: a pure version diff is empty, but a live // trigger is not actually registered, so it must be re-added. @@ -316,6 +350,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), + 'update', ); }); @@ -365,6 +400,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), + 'update', ); }); @@ -386,6 +422,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['b', 'c']), + 'update', ); }); @@ -453,6 +490,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), + 'update', ); // The cache is invalidated before the version is advanced and repopulated // straight after, so the empty window never serves a stale version, all @@ -664,6 +702,7 @@ describe('WorkflowPublicationApplier', () => { expect.objectContaining({ id: 'wf-1' }), newVersion, new Set(['a']), + 'update', ); }); }); 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 2e423efc2b2..c5e0e1cdb34 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 @@ -124,6 +124,40 @@ describe('WorkflowPublicationReconciler', () => { }); }); + describe('enqueue reason per pass', () => { + beforeEach(() => { + triggerStatusRepository.findActivatedInMemoryTriggers.mockResolvedValue([ + { workflowId: 'wf-1', nodeId: 'n1' }, + ]); + setRegistered({}); // n1 not registered → wf-1 is re-enqueued every pass + }); + + it("stamps 'startup' on the initial pass at boot", async () => { + service.init(); + await vi.advanceTimersByTimeAsync(0); + + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-1'], 'startup'); + }); + + it("stamps 'reconcile' on an interval tick", async () => { + service.startReconciler(); + await vi.advanceTimersByTimeAsync(5_000); + + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-1'], 'reconcile'); + }); + + it("stamps 'leadership-takeover' on the takeover pass", async () => { + service.startReconciler(); + + await service.reconcileOnLeaderTakeover(); + + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith( + ['wf-1'], + 'leadership-takeover', + ); + }); + }); + describe('loop lifecycle', () => { it('does not start when the publication service is disabled', async () => { Object.assign(config, { useWorkflowPublicationService: false }); @@ -207,9 +241,9 @@ describe('WorkflowPublicationReconciler', () => { ]); setRegistered({ 'wf-1': ['n1'] }); // n2 is missing - await service.reconcile(); + await service.reconcile('reconcile'); - expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-1']); + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-1'], 'reconcile'); expect(outboxConsumer.startPolling).toHaveBeenCalled(); expect(outboxConsumer.drainPending).toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( @@ -224,7 +258,7 @@ describe('WorkflowPublicationReconciler', () => { ]); setRegistered({ 'wf-1': ['n1'] }); - await service.reconcile(); + await service.reconcile('reconcile'); expect(outboxRepository.enqueueByWorkflowIds).not.toHaveBeenCalled(); expect(outboxConsumer.drainPending).not.toHaveBeenCalled(); @@ -237,17 +271,17 @@ describe('WorkflowPublicationReconciler', () => { ]); setRegistered({ 'wf-1': ['n1'], 'wf-2': [] }); // only wf-2 is missing a trigger - await service.reconcile(); + await service.reconcile('reconcile'); - expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-2']); + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-2'], 'reconcile'); }); it('re-enqueues a workflow whose published version diverged from the active version', async () => { outboxRepository.findVersionSkewedWorkflowIds.mockResolvedValue(['wf-skew']); - await service.reconcile(); + await service.reconcile('reconcile'); - expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-skew']); + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-skew'], 'reconcile'); expect(outboxConsumer.drainPending).toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( 'workflow-publication-reconciliation', @@ -258,9 +292,9 @@ describe('WorkflowPublicationReconciler', () => { it('re-enqueues a workflow whose trigger-status rows lag the active version', async () => { outboxRepository.findTriggerStatusDriftedWorkflowIds.mockResolvedValue(['wf-drift']); - await service.reconcile(); + await service.reconcile('reconcile'); - expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-drift']); + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-drift'], 'reconcile'); expect(outboxConsumer.drainPending).toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( 'workflow-publication-reconciliation', @@ -271,9 +305,12 @@ describe('WorkflowPublicationReconciler', () => { it('re-enqueues a published workflow that has no trigger-status rows', async () => { outboxRepository.findUnreportedPublishedWorkflowIds.mockResolvedValue(['wf-unreported']); - await service.reconcile(); + await service.reconcile('reconcile'); - expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith(['wf-unreported']); + expect(outboxRepository.enqueueByWorkflowIds).toHaveBeenCalledWith( + ['wf-unreported'], + 'reconcile', + ); expect(outboxConsumer.drainPending).toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( 'workflow-publication-reconciliation', @@ -286,7 +323,7 @@ describe('WorkflowPublicationReconciler', () => { new Error('DB error'), ); - await expect(service.reconcile()).resolves.toBeUndefined(); + await expect(service.reconcile('reconcile')).resolves.toBeUndefined(); expect(errorReporter.error).toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( 'workflow-publication-reconciliation', @@ -297,7 +334,7 @@ describe('WorkflowPublicationReconciler', () => { it('runs the ghost sweep and none of the leader detections when not leader', async () => { Object.assign(instanceSettings, { isLeader: false }); - await service.reconcile(); + await service.reconcile('reconcile'); expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalledTimes(1); expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled(); @@ -312,7 +349,7 @@ describe('WorkflowPublicationReconciler', () => { }); it('does not run the ghost sweep on the leader', async () => { - await service.reconcile(); + await service.reconcile('reconcile'); expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled(); expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled(); @@ -333,7 +370,7 @@ describe('WorkflowPublicationReconciler', () => { it('tears down ghost triggers under the workflow lock and reports the surplus', async () => { setGhost('wf-ghost'); - await service.reconcile(); + await service.reconcile('reconcile'); expect(lifecycleLock.runExclusive).toHaveBeenCalledWith('wf-ghost', expect.any(Function)); expect(activeWorkflowTriggers.remove).toHaveBeenCalledWith('wf-ghost'); @@ -347,7 +384,7 @@ describe('WorkflowPublicationReconciler', () => { activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds.mockReturnValue(['wf-1']); workflowRepository.getActiveIds.mockResolvedValue(['wf-1']); - await service.reconcile(); + await service.reconcile('reconcile'); expect(activeWorkflowTriggers.remove).not.toHaveBeenCalled(); expect(eventService.emit).toHaveBeenCalledWith( @@ -362,7 +399,7 @@ describe('WorkflowPublicationReconciler', () => { mock({ workflowId: 'wf-ghost' }), ); - await service.reconcile(); + await service.reconcile('reconcile'); expect(activeWorkflowTriggers.remove).not.toHaveBeenCalled(); }); @@ -376,7 +413,7 @@ describe('WorkflowPublicationReconciler', () => { return true; }); - await service.reconcile(); + await service.reconcile('reconcile'); expect(activeWorkflowTriggers.remove).toHaveBeenCalledWith('wf-good'); expect(errorReporter.error).toHaveBeenCalledWith(expect.any(Error), { @@ -400,7 +437,7 @@ describe('WorkflowPublicationReconciler', () => { activeVersionId: 'v-2', } as WorkflowEntity); - await service.reconcile(); + await service.reconcile('reconcile'); expect(activeWorkflowTriggers.remove).not.toHaveBeenCalled(); }); diff --git a/packages/cli/src/workflows/publication/workflow-publication-applier.ts b/packages/cli/src/workflows/publication/workflow-publication-applier.ts index 70f8147ecf4..32f475543c1 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-applier.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-applier.ts @@ -4,13 +4,14 @@ import { WorkflowHistory, WorkflowHistoryRepository, WorkflowPublicationOutbox, + WorkflowPublicationReason, WorkflowPublishedVersionRepository, WorkflowRepository, type WorkflowPublicationTriggerKind, } from '@n8n/db'; import { Service } from '@n8n/di'; import { ensureError } from '@n8n/utils/errors/ensure-error'; -import type { INode } from 'n8n-workflow'; +import type { INode, WorkflowActivateMode } from 'n8n-workflow'; import type { PublicationResult, @@ -24,6 +25,19 @@ import { } from '@/workflows/triggers/workflow-trigger-activator'; import { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service'; +/** + * The activation mode reported to trigger nodes for each enqueue reason, so + * e.g. the n8n Trigger's "Instance Started" event fires exactly for the + * leader's startup pass. Records from before the `reason` column existed + * default to `publish` at the DB level, i.e. today's `update` behavior. + */ +const ACTIVATION_MODE_BY_REASON: Record = { + [WorkflowPublicationReason.Publish]: 'update', + [WorkflowPublicationReason.Startup]: 'init', + [WorkflowPublicationReason.LeadershipTakeover]: 'leadershipChange', + [WorkflowPublicationReason.Reconcile]: 'update', +}; + /** * Reconciles a workflow's triggers to a published version, one outbox record at * a time. This is the only class that knows the remove → advance published @@ -152,7 +166,14 @@ export class WorkflowPublicationApplier { try { if (toAdd.size > 0) { - const outcome = await this.workflowTriggerActivator.activate(workflow, newVersion, toAdd); + const activationMode = + ACTIVATION_MODE_BY_REASON[record.reason ?? WorkflowPublicationReason.Publish]; + const outcome = await this.workflowTriggerActivator.activate( + workflow, + newVersion, + toAdd, + activationMode, + ); return this.classifyActivationOutcome(outcome, desiredTriggerNodes, triggerKinds); } 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 11d7541afd0..f9bbfbaced3 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-reconciler.service.ts @@ -3,6 +3,7 @@ import { WorkflowsConfig } from '@n8n/config'; import { Time } from '@n8n/constants'; import { WorkflowPublicationOutboxRepository, + WorkflowPublicationReason, WorkflowPublicationTriggerStatusRepository, WorkflowRepository, } from '@n8n/db'; @@ -87,7 +88,7 @@ export class WorkflowPublicationReconciler { // Run an initial pass at startup rather than waiting a full interval, so a // trigger missed during the last leader transition recovers promptly. - if (this.reconcileInterval) void this.reconcile(); + if (this.reconcileInterval) void this.reconcile(WorkflowPublicationReason.Startup); } /** @@ -101,7 +102,7 @@ export class WorkflowPublicationReconciler { const intervalSeconds = this.workflowsConfig.publicationReconcileIntervalSeconds; this.reconcileInterval = setInterval( - async () => await this.reconcile(), + async () => await this.reconcile(WorkflowPublicationReason.Reconcile), intervalSeconds * Time.seconds.toMilliseconds, ); @@ -122,8 +123,12 @@ export class WorkflowPublicationReconciler { * other role: sweep ghost registrations instead. Errors are caught and * reported so a transient failure never escapes the interval; the next pass * retries. + * + * `reason` records why this pass enqueues (startup, takeover, or drift-heal); + * the applier translates it into the activation mode reported to trigger + * nodes, e.g. "Instance Started" fires only for the startup pass. */ - async reconcile(): Promise { + async reconcile(reason: WorkflowPublicationReason): Promise { if (this.isShuttingDown) return; if (!this.instanceSettings.isLeader) { @@ -140,22 +145,26 @@ export class WorkflowPublicationReconciler { const missing = await this.republishWorkflows( await this.findMissingActiveWorkflows(), 'Re-publishing workflows with missing in-memory triggers', + reason, ); // After the missing-trigger drain, so workflows that pass already // converged are not re-detected (and double-counted) as skewed. const versionSkew = await this.republishWorkflows( await this.outboxRepository.findVersionSkewedWorkflowIds(), 'Re-enqueuing workflows whose published version diverged from the active version', + reason, ); // 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', + reason, ); const unreported = await this.republishWorkflows( await this.outboxRepository.findUnreportedPublishedWorkflowIds(), 'Re-enqueuing published workflows that no publication reported trigger statuses for', + reason, ); span.setAttribute('n8n.publication.deficient_workflows', missing); @@ -203,7 +212,7 @@ export class WorkflowPublicationReconciler { async reconcileOnLeaderTakeover(): Promise { if (!this.reconcileInterval) return; - await this.reconcile(); + await this.reconcile(WorkflowPublicationReason.LeadershipTakeover); } private async findSurplusWorkflowIds(): Promise { @@ -259,10 +268,14 @@ export class WorkflowPublicationReconciler { return missing; } - private async republishWorkflows(workflowIds: WorkflowId[], logMessage: string): Promise { + private async republishWorkflows( + workflowIds: WorkflowId[], + logMessage: string, + reason: WorkflowPublicationReason, + ): Promise { if (workflowIds.length > 0) { this.logger.debug(logMessage, { workflowIds }); - await this.outboxRepository.enqueueByWorkflowIds(workflowIds); + await this.outboxRepository.enqueueByWorkflowIds(workflowIds, reason); // Drain directly rather than waiting for the next poll cycle. These are // safe to call here, to recover more quickly. this.outboxConsumer.startPolling(); diff --git a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts index 893e2e20fa4..b6e2648fdae 100644 --- a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts @@ -285,6 +285,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['t', 'p', 'webhook-node']), + 'update', ); // Both phases overlap inside one isolate bracket, so their relative order is @@ -304,6 +305,48 @@ describe('WorkflowTriggerActivator', () => { expect(workflowRepository.updateWorkflowTriggerCount).toHaveBeenCalledWith('wf-1', 2); }); + test('threads the activation mode to both the non-webhook and webhook registrations', async () => { + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue( + mock(), + ); + + const webhookTriggerRegistrar = mock(); + const webhookData = mock({ node: 'Webhook' }); + webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([webhookData]); + + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.createRegistrationContext.mockReturnValue( + mock(), + ); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['t']); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const dbWorkflow = mock({ + id: 'wf-1', + name: 'Test workflow', + staticData: {}, + settings: {}, + }); + + await activator.activate( + dbWorkflow, + { + nodes: [node('t', 'trigger'), node('webhook-node', 'webhook', { name: 'Webhook' })], + connections: {}, + }, + new Set(['t', 'webhook-node']), + 'init', + ); + + expect(nonWebhookTriggerRegistrar.createRegistrationContext).toHaveBeenCalledWith( + dbWorkflow, + expect.objectContaining({ activationMode: 'init' }), + ); + expect(webhookTriggerRegistrar.register).toHaveBeenCalledWith( + expect.objectContaining({ webhookData, activation: 'init' }), + ); + }); + test('keeps the activation isolate until both concurrent phases settle after a phase error', async () => { const callOrder: string[] = []; vi.spyOn(WorkflowExpression.prototype, 'acquireIsolate').mockImplementation(async () => { @@ -347,6 +390,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-node', 'trigger-node']), + 'update', ), ).rejects.toThrow('webhook discovery failed'); @@ -503,6 +547,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-ok', 'webhook-bad', 't', 'p']), + 'update', ); // Both registrars ran; each phase surfaced its own failure while keeping the @@ -546,6 +591,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-a', 'webhook-b']), + 'update', ); // Parallel fan-out: assert by membership, not order. @@ -585,6 +631,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('webhook-a', 'webhook', { name: 'Webhook A' })], connections: {} }, new Set(['webhook-a']), + 'update', ); expect(outcome).toEqual({ activated: ['webhook-a'], failures: [] }); @@ -630,6 +677,7 @@ describe('WorkflowTriggerActivator', () => { connections: {}, }, new Set(['webhook-a', 'webhook-b']), + 'update', ); // Parallel fan-out: assert by membership, not order. @@ -670,6 +718,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} }, new Set(['webhook-node']), + 'update', ); expect(outcome.activated).toEqual([]); @@ -707,6 +756,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('t', 'trigger'), node('p', 'poll')], connections: {} }, new Set(['t', 'p']), + 'update', ); expect(outcome.activated).toEqual(expect.arrayContaining(['t'])); @@ -748,6 +798,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('p', 'poll')], connections: {} }, new Set(['p']), + 'update', ); expect(outcome).toEqual({ activated: ['p'], failures: [] }); @@ -787,6 +838,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('t', 'trigger')], connections: {} }, new Set(['t']), + 'update', ); const context = deps.nonWebhookTriggerRegistrar.createRegistrationContext.mock.calls[0][1]; @@ -929,6 +981,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('webhook-a', 'webhook', { name: 'Webhook A' })], connections: {} }, new Set(['webhook-a']), + 'update', ); expect(getEmissions('workflow-publication-trigger-operation')).toContainEqual( @@ -957,6 +1010,7 @@ describe('WorkflowTriggerActivator', () => { mock({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }), { nodes: [node('webhook-b', 'webhook', { name: 'Webhook B' })], connections: {} }, new Set(['webhook-b']), + 'update', ); expect(getEmissions('workflow-publication-trigger-operation')).toContainEqual( diff --git a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts index 7bf9357ec05..1565682da6a 100644 --- a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts +++ b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts @@ -244,10 +244,11 @@ export class WorkflowTriggerActivator { dbWorkflow: WorkflowEntity, version: WorkflowTriggerVersion, nodeIds: Set, + activationMode: WorkflowActivateMode, ): Promise { const startedAt = Date.now(); try { - const outcome = await this.activateInternal(dbWorkflow, version, nodeIds); + const outcome = await this.activateInternal(dbWorkflow, version, nodeIds, activationMode); this.emitTriggerOperation( 'activate', outcome.failures.length === 0 ? 'success' : 'failure', @@ -265,6 +266,7 @@ export class WorkflowTriggerActivator { dbWorkflow: WorkflowEntity, version: WorkflowTriggerVersion, nodeIds: Set, + activationMode: WorkflowActivateMode, ): Promise { return await this.tracing.startSpan( { @@ -294,7 +296,13 @@ export class WorkflowTriggerActivator { // The two phases share the single isolate acquired here and mutate // `outcome` via synchronous pushes, so overlapping them is safe. const phaseResults = await Promise.allSettled([ - this.registerWebhookTriggers(workflow, additionalData, nodeIds, outcome), + this.registerWebhookTriggers( + workflow, + additionalData, + nodeIds, + outcome, + activationMode, + ), this.registerNonWebhookTriggers( dbWorkflow, workflow, @@ -302,6 +310,7 @@ export class WorkflowTriggerActivator { resolveWorkflowData, nodeIds, outcome, + activationMode, ), ]); this.throwRejectedPhaseError(phaseResults); @@ -495,12 +504,19 @@ export class WorkflowTriggerActivator { additionalData: IWorkflowExecuteAdditionalData, nodeIds: Set, outcome: TriggerActivationOutcome, + activationMode: WorkflowActivateMode, ) { const webhooksByNode = this.groupWebhookTriggersByNode(workflow, additionalData, nodeIds); const tasks = [...webhooksByNode].map( async ([nodeId, { nodeName, webhooks }]) => - await this.registerWebhookTriggersForNode(workflow, nodeId, nodeName, webhooks), + await this.registerWebhookTriggersForNode( + workflow, + nodeId, + nodeName, + webhooks, + activationMode, + ), ); for (const result of await Promise.all(tasks)) { @@ -519,6 +535,7 @@ export class WorkflowTriggerActivator { nodeId: INode['id'], nodeName: string, webhooks: IWebhookData[], + activationMode: WorkflowActivateMode, ): Promise> { try { for (const webhookData of webhooks) { @@ -528,7 +545,7 @@ export class WorkflowTriggerActivator { workflow, webhookData, mode: 'trigger', - activation: 'update', + activation: activationMode, }), TRIGGER_ACTIVATION_MAX_ATTEMPTS, ); @@ -622,12 +639,13 @@ export class WorkflowTriggerActivator { resolveWorkflowData: () => Promise, nodeIds: Set, outcome: TriggerActivationOutcome, + activationMode: WorkflowActivateMode, ) { const triggerNodeIds = this.getNonWebhookTriggerNodeIdsForNodeIds(workflow, nodeIds); if (triggerNodeIds.length === 0) return; const registration = this.nonWebhookTriggerRegistrar.createRegistrationContext(dbWorkflow, { - activationMode: 'update', + activationMode, executionMode: 'trigger', additionalData, resolveWorkflowData, diff --git a/packages/cli/src/workflows/workflow.service.ts b/packages/cli/src/workflows/workflow.service.ts index 43bcbd697ab..5c741ba3de3 100644 --- a/packages/cli/src/workflows/workflow.service.ts +++ b/packages/cli/src/workflows/workflow.service.ts @@ -11,6 +11,7 @@ import { WorkflowRepository, WorkflowPublishHistoryRepository, WorkflowPublicationOutboxRepository, + WorkflowPublicationReason, WorkflowPublishedVersionRepository, ProjectRepository, } from '@n8n/db'; @@ -1583,7 +1584,12 @@ export class WorkflowService { trx, ); - await this.outboxRepository.enqueue(workflowId, versionIdToActivate, trx); + await this.outboxRepository.enqueue( + workflowId, + versionIdToActivate, + WorkflowPublicationReason.Publish, + trx, + ); }); // Wake the leader now that the record is committed, so it drains without @@ -1624,7 +1630,12 @@ export class WorkflowService { trx, ); - await this.outboxRepository.enqueue(workflowId, deactivatedVersionId, trx); + await this.outboxRepository.enqueue( + workflowId, + deactivatedVersionId, + WorkflowPublicationReason.Publish, + trx, + ); // Durable jobs are DB state, so their removal commits here rather than // waiting on the leader's outbox handler: a lost hand-off would otherwise diff --git a/packages/cli/test/integration/commands/start.cmd.test.ts b/packages/cli/test/integration/commands/start.cmd.test.ts index c4a904612b7..397408609f0 100644 --- a/packages/cli/test/integration/commands/start.cmd.test.ts +++ b/packages/cli/test/integration/commands/start.cmd.test.ts @@ -140,7 +140,7 @@ describe('Start.run() with workflow publication service', () => { ); const outboxRepository = Container.get(WorkflowPublicationOutboxRepository); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); // What a follower's publish triggers on this (leader) instance via Redis. // Re-emitted per retry: a wake-up that lands while a previous drain is 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 7dda751047c..c5516327aaa 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 @@ -22,7 +22,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('enqueues a pending record that can then be claimed', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); @@ -35,8 +35,8 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('supersedes an existing pending record when re-enqueued for the same workflow', async () => { - await repository.enqueue('wf-1', 'v-1'); - await repository.enqueue('wf-1', 'v-2'); + await repository.enqueue('wf-1', 'v-1', 'publish'); + await repository.enqueue('wf-1', 'v-2', 'publish'); const claimed = await repository.claimNextPendingRecord(); expect(claimed?.workflowId).toBe('wf-1'); @@ -47,9 +47,58 @@ describe('WorkflowPublicationOutboxRepository', () => { expect(claimedAgain).toBeNull(); }); + describe('reason', () => { + it('persists the reason it was enqueued with', async () => { + await repository.enqueue('wf-1', 'v-1', 'publish'); + + const claimed = await repository.claimNextPendingRecord(); + expect(claimed?.reason).toBe('publish'); + }); + + it('supersedes the reason of a pending record on re-enqueue, so a fresher intent wins', async () => { + const workflow = await createActiveWorkflow(); + assert(workflow.activeVersionId); + await repository.enqueueByWorkflowIds([workflow.id], 'startup'); + + // A user publish lands while the startup record is still pending. + await repository.enqueue(workflow.id, workflow.activeVersionId, 'publish'); + + const claimed = await repository.claimNextPendingRecord(); + expect(claimed?.workflowId).toBe(workflow.id); + expect(claimed?.reason).toBe('publish'); + }); + + it('keeps a pending record untouched when the reconciler re-enqueues over it', async () => { + const workflow = await createActiveWorkflow(); + assert(workflow.activeVersionId); + await repository.enqueue(workflow.id, workflow.activeVersionId, 'publish'); + + // The reconciler's bulk enqueue conflicts with the pending user record + // and must not overwrite it (DO NOTHING). + await repository.enqueueByWorkflowIds([workflow.id], 'startup'); + + const claimed = await repository.claimNextPendingRecord(); + expect(claimed?.workflowId).toBe(workflow.id); + expect(claimed?.reason).toBe('publish'); + }); + + it('stamps the given reason on bulk-enqueued records', async () => { + const wf1 = await createActiveWorkflow(); + const wf2 = await createActiveWorkflow(); + + await repository.enqueueByWorkflowIds([wf1.id, wf2.id], 'leadership-takeover'); + + const pending = await repository.find({ where: { status: 'pending' } }); + expect(pending).toHaveLength(2); + for (const record of pending) { + expect(record.reason).toBe('leadership-takeover'); + } + }); + }); + it('enqueues within a provided transaction and is visible once it commits', async () => { await repository.manager.transaction(async (trx) => { - await repository.enqueue('wf-1', 'v-1', trx); + await repository.enqueue('wf-1', 'v-1', 'publish', trx); }); const claimed = await repository.claimNextPendingRecord(); @@ -60,7 +109,7 @@ describe('WorkflowPublicationOutboxRepository', () => { it('discards the enqueued record when the surrounding transaction rolls back', async () => { await expect( repository.manager.transaction(async (trx) => { - await repository.enqueue('wf-1', 'v-1', trx); + await repository.enqueue('wf-1', 'v-1', 'publish', trx); throw new Error('rollback'); }), ).rejects.toThrow('rollback'); @@ -69,8 +118,8 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('claims pending records in FIFO order', async () => { - await repository.enqueue('wf-1', 'v-1'); - await repository.enqueue('wf-2', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); + await repository.enqueue('wf-2', 'v-1', 'publish'); const first = await repository.claimNextPendingRecord(); const second = await repository.claimNextPendingRecord(); @@ -81,14 +130,14 @@ describe('WorkflowPublicationOutboxRepository', () => { it('does not claim a second record for a workflow already in progress', async () => { // wf-1 is claimed (in progress), then a newer version is enqueued. - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const inProgress = await repository.claimNextPendingRecord(); assert(inProgress); - await repository.enqueue('wf-1', 'v-2'); + await repository.enqueue('wf-1', 'v-2', 'publish'); // A different workflow is claimable, but wf-1's new pending record is not // until its in-progress record is resolved. - await repository.enqueue('wf-2', 'v-1'); + await repository.enqueue('wf-2', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); expect(claimed?.workflowId).toBe('wf-2'); expect(await repository.claimNextPendingRecord()).toBeNull(); @@ -101,12 +150,12 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('enqueues a fresh pending record once the previous one is no longer pending', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); await repository.markCompleted(claimed.id); - await repository.enqueue('wf-1', 'v-2'); + await repository.enqueue('wf-1', 'v-2', 'publish'); const next = await repository.claimNextPendingRecord(); expect(next?.id).not.toBe(claimed.id); @@ -114,7 +163,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('marks a claimed record as completed', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); @@ -126,7 +175,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('marks a claimed record as failed and records the error', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); @@ -138,7 +187,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('throws when transitioning a record that is not in progress', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); await repository.markCompleted(claimed.id); @@ -150,7 +199,7 @@ describe('WorkflowPublicationOutboxRepository', () => { describe('returnToPending', () => { it('returns a claimed record to the queue so it can be claimed again', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); @@ -166,10 +215,10 @@ describe('WorkflowPublicationOutboxRepository', () => { it('drops the claimed record when a newer pending record already supersedes it', async () => { // wf-1 is claimed (in progress), then a newer version is enqueued as pending. - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); - await repository.enqueue('wf-1', 'v-2'); + await repository.enqueue('wf-1', 'v-2', 'publish'); await repository.returnToPending(claimed.id); @@ -181,7 +230,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('is a no-op when the record is no longer in progress', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); await repository.markCompleted(claimed.id); @@ -215,7 +264,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('reclaims a stale in_progress record', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); await backdateUpdatedAt(claimed.id); @@ -227,7 +276,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('does not reclaim a fresh in_progress record', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); @@ -236,7 +285,7 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('bumps updatedAt on reclaim so it is not immediately reclaimable again', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); await backdateUpdatedAt(claimed.id); @@ -249,10 +298,10 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('reclaims the stale in_progress only, leaving a newer pending record untouched', async () => { - await repository.enqueue('wf-1', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); - await repository.enqueue('wf-1', 'v-2'); + await repository.enqueue('wf-1', 'v-2', 'publish'); await backdateUpdatedAt(claimed.id); const reclaimed = await repository.claimNextPendingRecord(); @@ -282,7 +331,7 @@ describe('WorkflowPublicationOutboxRepository', () => { workflowId: string, outcome: 'completed' | 'failed' | 'partial', ) => { - await repository.enqueue(workflowId, 'v-1'); + await repository.enqueue(workflowId, 'v-1', 'publish'); const claimed = await repository.claimNextPendingRecord(); assert(claimed); if (outcome === 'completed') await repository.markCompleted(claimed.id); @@ -335,8 +384,8 @@ describe('WorkflowPublicationOutboxRepository', () => { }); it('never deletes pending or in_progress rows', async () => { - await repository.enqueue('wf-1', 'v-1'); // stays pending - await repository.enqueue('wf-2', 'v-1'); + await repository.enqueue('wf-1', 'v-1', 'publish'); // stays pending + await repository.enqueue('wf-2', 'v-1', 'publish'); const inProgress = await repository.claimNextPendingRecord(); assert(inProgress); await backdateUpdatedAt(inProgress.id); @@ -385,7 +434,7 @@ describe('WorkflowPublicationOutboxRepository', () => { const wf2 = await createActiveWorkflow(); const wf3 = await createActiveWorkflow(); - await repository.enqueueByWorkflowIds([wf1.id, wf3.id]); + await repository.enqueueByWorkflowIds([wf1.id, wf3.id], 'reconcile'); const pending = await repository.find({ where: { status: 'pending' } }); expect(pending).toEqual( @@ -407,7 +456,7 @@ describe('WorkflowPublicationOutboxRepository', () => { const archived = await createWorkflow(); await Container.get(WorkflowRepository).update(archived.id, { isArchived: true }); - await repository.enqueueByWorkflowIds([unpublished.id, archived.id]); + await repository.enqueueByWorkflowIds([unpublished.id, archived.id], 'reconcile'); const pending = await repository.find({ where: { status: 'pending' } }); expect(pending).toEqual( @@ -428,8 +477,8 @@ describe('WorkflowPublicationOutboxRepository', () => { it('is idempotent: re-running does not create duplicate pending records', async () => { const workflow = await createActiveWorkflow(); - await repository.enqueueByWorkflowIds([workflow.id]); - await repository.enqueueByWorkflowIds([workflow.id]); + await repository.enqueueByWorkflowIds([workflow.id], 'reconcile'); + await repository.enqueueByWorkflowIds([workflow.id], 'reconcile'); const pending = await repository.find({ where: { workflowId: workflow.id, status: 'pending' }, @@ -444,9 +493,9 @@ describe('WorkflowPublicationOutboxRepository', () => { // record is at least as fresh as reconciliation's snapshot and must win — // overwriting it could roll the workflow back to a stale version. const workflow = await createActiveWorkflow(); - await repository.enqueue(workflow.id, 'v-concurrent'); + await repository.enqueue(workflow.id, 'v-concurrent', 'publish'); - await repository.enqueueByWorkflowIds([workflow.id]); + await repository.enqueueByWorkflowIds([workflow.id], 'reconcile'); const pending = await repository.find({ where: { workflowId: workflow.id, status: 'pending' }, @@ -458,7 +507,7 @@ describe('WorkflowPublicationOutboxRepository', () => { it('is a no-op for an empty list', async () => { await createActiveWorkflow(); - await expect(repository.enqueueByWorkflowIds([])).resolves.toBeUndefined(); + await expect(repository.enqueueByWorkflowIds([], 'reconcile')).resolves.toBeUndefined(); expect(await repository.find({ where: { status: 'pending' } })).toHaveLength(0); }); @@ -477,7 +526,7 @@ describe('WorkflowPublicationOutboxRepository', () => { ); } - await repository.enqueueByWorkflowIds(workflowIds); + await repository.enqueueByWorkflowIds(workflowIds, 'reconcile'); 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 8b844ac19ba..355fa4ee8d8 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 @@ -121,7 +121,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { connections: {}, }); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); expect(record).not.toBeNull(); @@ -162,7 +162,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { expect(presentResponse).toBeDefined(); expect(activeWorkflowTriggers.get(workflow.id)?.has(missing.id)).toBe(false); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); @@ -189,7 +189,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { const responseBefore = activeWorkflowTriggers.get(workflow.id)?.get(trigger.id); expect(responseBefore).toBeDefined(); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); @@ -223,7 +223,7 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => { connections: {}, }); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); 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 0dea8c2743c..0cc373316ea 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts @@ -121,7 +121,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // Publish through the real pipeline so the trigger registers in memory and // the reporter persists the `activated` trigger-status rows with kinds. - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); @@ -135,7 +135,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // One reconcile pass detects the deficit, re-enqueues, and drains: the // applier re-registers the missing trigger at the unchanged version. - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); const published = await publishedVersionRepository.getPublishedVersionWithRelations( @@ -145,6 +145,47 @@ describe('WorkflowPublicationReconciler (integration)', () => { expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); }); + test('a startup pass registers recovered triggers with the init activation mode', async () => { + const owner = await createOwner(); + + const trigger = scheduleNode('boot'); + const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); + await setActiveVersion(workflow.id, workflow.versionId); + + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); + const record = await outboxRepository.claimNextPendingRecord(); + await consumer.processRecord(record!); + + // A fresh boot: the registry is empty, the published state is intact. + await activeWorkflowTriggers.remove(workflow.id); + + const addTriggersSpy = vi.spyOn(activeWorkflowTriggers, 'addTriggers'); + await reconciler.reconcile('startup'); + + expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); + // The trigger node was told the instance just started, so events like the + // n8n Trigger's "Instance Started" fire for every workflow of the pass. + expect(addTriggersSpy).toHaveBeenCalledWith( + workflow.id, + expect.anything(), + [trigger.id], + expect.anything(), + 'trigger', + 'init', + expect.anything(), + expect.anything(), + ); + + const [terminalRecord] = await outboxRepository.find({ + where: { workflowId: workflow.id, status: 'completed' }, + order: { id: 'DESC' }, + take: 1, + }); + expect(terminalRecord.reason).toBe('startup'); + + addTriggersSpy.mockRestore(); + }); + test('clears orphaned trigger-status rows of an unpublished workflow by re-running the unpublish', async () => { const owner = await createOwner(); @@ -154,7 +195,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // Publish through the real pipeline so the reporter persists the // `activated` trigger-status rows. - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); @@ -170,7 +211,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // One reconcile pass surfaces the orphaned rows as a deficit, enqueues the // workflow, and the drained unpublish clears the rows and completes. - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(await triggerStatusRepository.findByWorkflowId(workflow.id)).toHaveLength(0); expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined(); @@ -184,7 +225,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); @@ -197,7 +238,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { await publishedVersionRepository.removePublishedVersion(workflow.id); await triggerStatusRepository.delete({ workflowId: workflow.id }); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined(); // Repair is local — no outbox round-trip was needed. @@ -211,7 +252,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); @@ -223,7 +264,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // One pass converges: surplus teardown first, which lets the leftover rows // read as missing, enqueue, and clear through the unpublish path. - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined(); expect(await triggerStatusRepository.findByWorkflowId(workflow.id)).toHaveLength(0); @@ -237,16 +278,16 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); // Mid-unpublish: activeVersionId already cleared, pending record owns the // teardown. Reconciliation must not race it. await Container.get(WorkflowRepository).update(workflow.id, { activeVersionId: null }); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); }); @@ -258,7 +299,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // A parameter-only newer version is published: the trigger node set is @@ -269,7 +310,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { nodes: [{ ...trigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }], }); await setActiveVersion(workflow.id, newVersionId); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBe(newVersionId); @@ -278,7 +319,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // and the running trigger's node id still matches the desired one. await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); // The skew check enqueued a third record and the drain converged the // mapping back to the active version. @@ -294,7 +335,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // An unpublish fully applied elsewhere (triggers down, status rows @@ -308,7 +349,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { workflow.versionId, ); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBeNull(); expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); @@ -321,7 +362,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // Mid-flight publish of a parameter-only new version: `activeVersionId` @@ -333,11 +374,11 @@ describe('WorkflowPublicationReconciler (integration)', () => { nodes: [{ ...trigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }], }); await setActiveVersion(workflow.id, newVersionId); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); expect(await outboxRepository.findVersionSkewedWorkflowIds()).not.toContain(workflow.id); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); // Untouched: the mapping still points at the old version and the pending // record is still pending — reconciliation neither enqueued nor drained. @@ -358,7 +399,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); const newVersionId = 'version-2-status-drift'; @@ -367,7 +408,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { nodes: [{ ...trigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }], }); await setActiveVersion(workflow.id, newVersionId); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // A zombie writer rewrites the status rows for the old version after its @@ -379,7 +420,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { ); expect(await outboxRepository.findVersionSkewedWorkflowIds()).not.toContain(workflow.id); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); const rows = await triggerStatusRepository.findByWorkflowId(workflow.id); expect(rows).toHaveLength(1); @@ -398,7 +439,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // while the flag was off has none, and no outbox record either. await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true); const rows = await triggerStatusRepository.findByWorkflowId(workflow.id); @@ -419,11 +460,11 @@ describe('WorkflowPublicationReconciler (integration)', () => { // 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); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await outboxRepository.markFailed(record!.id, 'unexpected error before reporting'); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); expect(await triggerStatusRepository.findByWorkflowId(workflow.id)).toHaveLength(0); @@ -436,7 +477,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // A v2 publish that crashed after advancing the mapping but before the @@ -451,11 +492,11 @@ describe('WorkflowPublicationReconciler (integration)', () => { }); await setActiveVersion(workflow.id, newVersionId); await publishedVersionRepository.setPublishedVersion(workflow.id, newVersionId); - await outboxRepository.enqueue(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await outboxRepository.markFailed(record!.id, 'unexpected error before reporting'); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); const rows = await triggerStatusRepository.findByWorkflowId(workflow.id); @@ -469,7 +510,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { 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 outboxRepository.enqueue(drifted.id, drifted.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); const newVersionId = 'version-2-drift-in-flight'; await createWorkflowHistory(drifted, owner, undefined, { @@ -477,7 +518,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { nodes: [{ ...driftTrigger, parameters: { rule: { interval: [{ field: 'hours' }] } } }], }); await setActiveVersion(drifted.id, newVersionId); - await outboxRepository.enqueue(drifted.id, newVersionId); + await outboxRepository.enqueue(drifted.id, newVersionId, 'publish'); // No rows yet, but the first publish is still pending. const unreportedTrigger = scheduleNode('unreported-in-flight'); @@ -486,14 +527,14 @@ describe('WorkflowPublicationReconciler (integration)', () => { owner, ); await setActiveVersion(unreported.id, unreported.versionId); - await outboxRepository.enqueue(unreported.id, unreported.versionId); + await outboxRepository.enqueue(unreported.id, unreported.versionId, 'publish'); expect(await outboxRepository.findTriggerStatusDriftedWorkflowIds()).not.toContain(drifted.id); expect(await outboxRepository.findUnreportedPublishedWorkflowIds()).not.toContain( unreported.id, ); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); // Untouched: both records are still pending — reconciliation neither // enqueued nor drained. @@ -510,7 +551,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!); // Activation is untouched: a genuine publish still registers the no-op @@ -529,7 +570,7 @@ describe('WorkflowPublicationReconciler (integration)', () => { // 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 reconciler.reconcile(); + await reconciler.reconcile('reconcile'); expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined(); }); @@ -541,13 +582,13 @@ describe('WorkflowPublicationReconciler (integration)', () => { const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); await setActiveVersion(workflow.id, workflow.versionId); - await outboxRepository.enqueue(workflow.id, workflow.versionId); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); const record = await outboxRepository.claimNextPendingRecord(); await consumer.processRecord(record!); const registeredBefore = activeWorkflowTriggers.get(workflow.id)?.get(trigger.id); expect(registeredBefore).toBeDefined(); - await reconciler.reconcile(); + await reconciler.reconcile('reconcile'); // Same response object: the trigger was left untouched, not re-registered. expect(activeWorkflowTriggers.get(workflow.id)?.get(trigger.id)).toBe(registeredBefore); diff --git a/packages/cli/test/migration/1786519946974-add-reason-to-workflow-publication-outbox.test.ts b/packages/cli/test/migration/1786519946974-add-reason-to-workflow-publication-outbox.test.ts new file mode 100644 index 00000000000..1376a04776f --- /dev/null +++ b/packages/cli/test/migration/1786519946974-add-reason-to-workflow-publication-outbox.test.ts @@ -0,0 +1,114 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + type TestMigrationContext, + undoLastSingleMigration, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; + +const MIGRATION_NAME = 'AddReasonToWorkflowPublicationOutbox1786519946974'; + +describe('AddReasonToWorkflowPublicationOutbox Migration', () => { + let dataSource: DataSource; + + async function insertRecord( + context: TestMigrationContext, + workflowId: string, + reason?: string, + ): Promise { + const table = context.escape.tableName('workflow_publication_outbox'); + if (reason === undefined) { + await context.runQuery( + `INSERT INTO ${table} ("workflowId", "publishedVersionId", "status") + VALUES (:workflowId, :versionId, :status)`, + { workflowId, versionId: 'v-1', status: 'pending' }, + ); + return; + } + await context.runQuery( + `INSERT INTO ${table} ("workflowId", "publishedVersionId", "status", "reason") + VALUES (:workflowId, :versionId, :status, :reason)`, + { workflowId, versionId: 'v-1', status: 'pending', reason }, + ); + } + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + dataSource = Container.get(DataSource); + }); + + // Each test starts from a clean database migrated up to (but not including) + // this migration and owns its fixtures, so the tests run in any order or + // in isolation. + beforeEach(async () => { + const context = createTestMigrationContext(dataSource); + await context.queryRunner.clearDatabase(); + await context.queryRunner.release(); + + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + it('adds the reason column with a publish default and enforces the enum check', async () => { + await runSingleMigration(MIGRATION_NAME); + + const context = createTestMigrationContext(dataSource); + try { + // A pre-upgrade writer that doesn't know the column gets the default. + await insertRecord(context, 'wf-default'); + const rows = await context.runQuery>( + `SELECT ${context.escape.columnName('reason')} FROM ${context.escape.tableName('workflow_publication_outbox')} + WHERE ${context.escape.columnName('workflowId')} = 'wf-default'`, + ); + expect(rows[0].reason).toBe('publish'); + + await expect(insertRecord(context, 'wf-startup', 'startup')).resolves.not.toThrow(); + await expect(insertRecord(context, 'wf-bogus', 'not-a-reason')).rejects.toThrow(); + } finally { + await context.queryRunner.release(); + } + }); + + it('rolls back cleanly, preserving rows, and can be reapplied', async () => { + await runSingleMigration(MIGRATION_NAME); + + let context = createTestMigrationContext(dataSource); + try { + await insertRecord(context, 'wf-defaulted'); + await insertRecord(context, 'wf-stamped', 'startup'); + } finally { + await context.queryRunner.release(); + } + + await undoLastSingleMigration(); + + context = createTestMigrationContext(dataSource); + try { + const rows = await context.runQuery>( + `SELECT ${context.escape.columnName('workflowId')} FROM ${context.escape.tableName('workflow_publication_outbox')}`, + ); + expect(rows.map(({ workflowId }) => workflowId).sort()).toEqual([ + 'wf-defaulted', + 'wf-stamped', + ]); + } finally { + await context.queryRunner.release(); + } + + await runSingleMigration(MIGRATION_NAME); + + context = createTestMigrationContext(dataSource); + try { + await expect(insertRecord(context, 'wf-after-redo', 'reconcile')).resolves.not.toThrow(); + } finally { + await context.queryRunner.release(); + } + }); +});