mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(core): Report the real activation mode for triggers activated via the publication outbox (#36123)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Mike Repeć
GitHub
parent
7f3713702a
commit
1a12af601b
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```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')))
|
||||
```
|
||||
|
||||
</details>
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
}
|
||||
}
|
||||
|
|
@ -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<WorkflowPublicationOutbox> {
|
||||
|
|
@ -47,23 +48,33 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
*
|
||||
* Pass `trx` to run the UPSERT inside an existing transaction, e.g. to make
|
||||
* the enqueue atomic with a `workflow_entity` update.
|
||||
*
|
||||
* The conflict path also supersedes `reason`: a fresher enqueue's intent
|
||||
* wins (e.g. a user publish replacing a pending startup record).
|
||||
*/
|
||||
async enqueue(
|
||||
workflowId: string,
|
||||
publishedVersionId: string,
|
||||
reason: WorkflowPublicationReason,
|
||||
trx?: EntityManager,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
const tableName = this.getTableName('workflow_publication_outbox');
|
||||
|
|
@ -71,27 +82,28 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
// `createdAt`/`updatedAt` carry DB-level defaults, so the insert omits
|
||||
// them; the conflict path bumps `updatedAt` explicitly.
|
||||
await trx.query(
|
||||
`INSERT INTO ${tableName} ("workflowId", "publishedVersionId", "status")
|
||||
VALUES ($1, $2, '${Status.Pending}')
|
||||
`INSERT INTO ${tableName} ("workflowId", "publishedVersionId", "status", "reason")
|
||||
VALUES ($1, $2, '${Status.Pending}', $3)
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO UPDATE SET "publishedVersionId" = EXCLUDED."publishedVersionId", "updatedAt" = CURRENT_TIMESTAMP(3)`,
|
||||
[workflowId, publishedVersionId],
|
||||
DO UPDATE SET "publishedVersionId" = EXCLUDED."publishedVersionId", "reason" = EXCLUDED."reason", "updatedAt" = CURRENT_TIMESTAMP(3)`,
|
||||
[workflowId, publishedVersionId, reason],
|
||||
);
|
||||
}
|
||||
|
||||
private async enqueueWithSqliteUpsert(
|
||||
workflowId: string,
|
||||
publishedVersionId: string,
|
||||
reason: WorkflowPublicationReason,
|
||||
trx: EntityManager,
|
||||
): Promise<void> {
|
||||
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<WorkflowPubl
|
|||
* an unpublish record that clears their stale trigger-status rows. Idempotent
|
||||
* via the same partial-unique-index upsert as {@link enqueue}.
|
||||
*/
|
||||
async enqueueByWorkflowIds(workflowIds: string[]): Promise<void> {
|
||||
async enqueueByWorkflowIds(
|
||||
workflowIds: string[],
|
||||
reason: WorkflowPublicationReason,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
private async enqueueByWorkflowIdsWithPostgresUpsert(
|
||||
workflowIds: string[],
|
||||
reason: WorkflowPublicationReason,
|
||||
): Promise<void> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
const workflowTableName = this.getTableName('workflow_entity');
|
||||
|
||||
|
|
@ -131,17 +149,20 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
// earlier snapshot). Such a record is at least as fresh as this statement's
|
||||
// snapshot — overwriting it could roll the workflow back to a stale version.
|
||||
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}', $2
|
||||
FROM ${workflowTableName} w
|
||||
WHERE w."id" = ANY($1)
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO NOTHING`,
|
||||
[workflowIds],
|
||||
[workflowIds, reason],
|
||||
);
|
||||
}
|
||||
|
||||
private async enqueueByWorkflowIdsWithSqliteUpsert(workflowIds: string[]): Promise<void> {
|
||||
private async enqueueByWorkflowIdsWithSqliteUpsert(
|
||||
workflowIds: string[],
|
||||
reason: WorkflowPublicationReason,
|
||||
): Promise<void> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
const workflowTableName = this.getTableName('workflow_entity');
|
||||
|
||||
|
|
@ -154,13 +175,13 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
const placeholders = chunk.map(() => '?').join(', ');
|
||||
|
||||
await this.query(
|
||||
`INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status")
|
||||
SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}'
|
||||
`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],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<WorkflowPublicationOutbox>({ 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();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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, WorkflowActivateMode> = {
|
||||
[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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
async reconcile(reason: WorkflowPublicationReason): Promise<void> {
|
||||
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<void> {
|
||||
if (!this.reconcileInterval) return;
|
||||
|
||||
await this.reconcile();
|
||||
await this.reconcile(WorkflowPublicationReason.LeadershipTakeover);
|
||||
}
|
||||
|
||||
private async findSurplusWorkflowIds(): Promise<WorkflowId[]> {
|
||||
|
|
@ -259,10 +268,14 @@ export class WorkflowPublicationReconciler {
|
|||
return missing;
|
||||
}
|
||||
|
||||
private async republishWorkflows(workflowIds: WorkflowId[], logMessage: string): Promise<number> {
|
||||
private async republishWorkflows(
|
||||
workflowIds: WorkflowId[],
|
||||
logMessage: string,
|
||||
reason: WorkflowPublicationReason,
|
||||
): Promise<number> {
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<IWorkflowExecuteAdditionalData>(),
|
||||
);
|
||||
|
||||
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
|
||||
const webhookData = mock<IWebhookData>({ node: 'Webhook' });
|
||||
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([webhookData]);
|
||||
|
||||
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
|
||||
nonWebhookTriggerRegistrar.createRegistrationContext.mockReturnValue(
|
||||
mock<PreparedNonWebhookTriggerRegistration>(),
|
||||
);
|
||||
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['t']);
|
||||
|
||||
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
|
||||
const dbWorkflow = mock<WorkflowEntity>({
|
||||
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<WorkflowEntity>({ 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<WorkflowEntity>({ 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<WorkflowEntity>({ 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<WorkflowEntity>({ 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<WorkflowEntity>({ 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<WorkflowEntity>({ 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<WorkflowEntity>({ 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(
|
||||
|
|
|
|||
|
|
@ -244,10 +244,11 @@ export class WorkflowTriggerActivator {
|
|||
dbWorkflow: WorkflowEntity,
|
||||
version: WorkflowTriggerVersion,
|
||||
nodeIds: Set<INode['id']>,
|
||||
activationMode: WorkflowActivateMode,
|
||||
): Promise<TriggerActivationOutcome> {
|
||||
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<INode['id']>,
|
||||
activationMode: WorkflowActivateMode,
|
||||
): Promise<TriggerActivationOutcome> {
|
||||
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<INode['id']>,
|
||||
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<Result<{ nodeId: INode['id'] }, TriggerActivationFailure>> {
|
||||
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<IWorkflowBase>,
|
||||
nodeIds: Set<INode['id']>,
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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!);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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<Array<{ reason: string }>>(
|
||||
`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<Array<{ workflowId: string }>>(
|
||||
`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();
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user