github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 21:49:36 +02:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity Graph

feat(core): add trigger seat and runner tables with leased claims

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Michael Siega 2026-08-31 13:54:22 +02:00
parent 19582fb20d
commit ffb2ec9b05
14 changed files with 1082 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
docs/generated/postgres-schema/README.md generated
View File

@ -115,6 +115,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.test_case_execution](public.test_case_execution.md) | 14 | | BASE TABLE |
| [public.test_run](public.test_run.md) | 16 | | BASE TABLE |
| [public.token_exchange_jti](public.token_exchange_jti.md) | 3 | | BASE TABLE |
| [public.trigger_runner](public.trigger_runner.md) | 4 | | BASE TABLE |
| [public.trusted_key](public.trusted_key.md) | 4 | | BASE TABLE |
| [public.trusted_key_source](public.trusted_key_source.md) | 8 | | BASE TABLE |
| [public.user](public.user.md) | 15 | | BASE TABLE |
@ -138,6 +139,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.workflow_review_request_workflow](public.workflow_review_request_workflow.md) | 5 | | BASE TABLE |
| [public.workflow_statistics](public.workflow_statistics.md) | 7 | | BASE TABLE |
| [public.workflow_statistics_delta](public.workflow_statistics_delta.md) | 6 | | BASE TABLE |
| [public.workflow_trigger_seat](public.workflow_trigger_seat.md) | 15 | | BASE TABLE |
| [public.workflows_tags](public.workflows_tags.md) | 2 | | BASE TABLE |
## Stored procedures and functions
@ -1413,6 +1415,12 @@ erDiagram
timestamp_3__with_time_zone expiresAt
varchar_255_ jti
}
"public.trigger_runner" {
timestamp_3__with_time_zone createdAt
timestamp_3__with_time_zone lastHeartbeatAt
varchar_255_ runnerId
timestamp_3__with_time_zone updatedAt
}
"public.trusted_key" {
timestamp_3__with_time_zone createdAt
text data
@ -1634,6 +1642,23 @@ erDiagram
varchar_36_ workflowId
varchar_128_ workflowName
}
"public.workflow_trigger_seat" {
varchar_16_ actualState
varchar_36_ actualVersionId
timestamp_3__with_time_zone createdAt
varchar_255_ desiredHolderId
varchar_16_ desiredState
varchar_36_ desiredVersionId
varchar_255_ holderId
bigint id
text lastError
integer leaseEpoch
timestamp_3__with_time_zone leaseExpiresAt
varchar_36_ nodeId
integer seatIndex
timestamp_3__with_time_zone updatedAt
varchar_36_ workflowId
}
"public.workflows_tags" {
varchar_36_ tagId FK
varchar_36_ workflowId FK

44
docs/generated/postgres-schema/public.trigger_runner.md generated Normal file
View File

@ -0,0 +1,44 @@
# public.trigger_runner
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| lastHeartbeatAt | timestamp(3) with time zone | | false | | | Upserted every reconcile tick; older than TTL means the runner is gone. |
| runnerId | varchar(255) | | false | | | |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| PK_f10c2f1c4aa12143266d0fa4db3 | PRIMARY KEY | PRIMARY KEY ("runnerId") |
| trigger_runner_createdAt_not_null | n | NOT NULL "createdAt" |
| trigger_runner_lastHeartbeatAt_not_null | n | NOT NULL "lastHeartbeatAt" |
| trigger_runner_runnerId_not_null | n | NOT NULL "runnerId" |
| trigger_runner_updatedAt_not_null | n | NOT NULL "updatedAt" |
## Indexes
| Name | Definition |
| ---- | ---------- |
| PK_f10c2f1c4aa12143266d0fa4db3 | CREATE UNIQUE INDEX "PK_f10c2f1c4aa12143266d0fa4db3" ON public.trigger_runner USING btree ("runnerId") |
## Relations
```mermaid
erDiagram
"public.trigger_runner" {
timestamp_3__with_time_zone createdAt
timestamp_3__with_time_zone lastHeartbeatAt
varchar_255_ runnerId
timestamp_3__with_time_zone updatedAt
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

75
docs/generated/postgres-schema/public.workflow_trigger_seat.md generated Normal file
View File

@ -0,0 +1,75 @@
# public.workflow_trigger_seat
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| actualState | varchar(16) | | true | | | What the holder last reported doing with the seat. |
| actualVersionId | varchar(36) | | true | | | The version the holder’s registered trigger is actually running. |
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| desiredHolderId | varchar(255) | | true | | | Handoff request: an underloaded runner asking the holder to release. |
| desiredState | varchar(16) | 'active'::character varying | false | | | Whether a runner should serve this seat; publication flips it instead of deleting rows so holders tear down cleanly. |
| desiredVersionId | varchar(36) | | false | | | The published version the seat’s trigger should run. |
| holderId | varchar(255) | | true | | | Runner currently holding the lease; NULL when vacant. |
| id | bigint | | false | | | |
| lastError | text | | true | | | |
| leaseEpoch | integer | 0 | false | | | Fencing token bumped on every claim; execution inserts are guarded on it so a stale holder’s emissions never land. |
| leaseExpiresAt | timestamp(3) with time zone | | true | | | When the current lease may be reclaimed by another runner. |
| nodeId | varchar(36) | | false | | | |
| seatIndex | integer | | false | | | Which of the trigger node’s N replica slots this row is. |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
| workflowId | varchar(36) | | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| CHK_workflow_trigger_seat_actualState | CHECK | CHECK ((("actualState")::text = ANY ((ARRAY['registered'::character varying, 'closed'::character varying, 'error'::character varying])::text[]))) |
| CHK_workflow_trigger_seat_desiredState | CHECK | CHECK ((("desiredState")::text = ANY ((ARRAY['active'::character varying, 'inactive'::character varying])::text[]))) |
| PK_d88e285cade3e5f2bb626a071f1 | PRIMARY KEY | PRIMARY KEY (id) |
| workflow_trigger_seat_createdAt_not_null | n | NOT NULL "createdAt" |
| workflow_trigger_seat_desiredState_not_null | n | NOT NULL "desiredState" |
| workflow_trigger_seat_desiredVersionId_not_null | n | NOT NULL "desiredVersionId" |
| workflow_trigger_seat_id_not_null | n | NOT NULL id |
| workflow_trigger_seat_leaseEpoch_not_null | n | NOT NULL "leaseEpoch" |
| workflow_trigger_seat_nodeId_not_null | n | NOT NULL "nodeId" |
| workflow_trigger_seat_seatIndex_not_null | n | NOT NULL "seatIndex" |
| workflow_trigger_seat_updatedAt_not_null | n | NOT NULL "updatedAt" |
| workflow_trigger_seat_workflowId_not_null | n | NOT NULL "workflowId" |
## Indexes
| Name | Definition |
| ---- | ---------- |
| IDX_workflow_trigger_seat_desiredState | CREATE INDEX "IDX_workflow_trigger_seat_desiredState" ON public.workflow_trigger_seat USING btree ("desiredState") |
| IDX_workflow_trigger_seat_workflowId_nodeId_seatIndex | CREATE UNIQUE INDEX "IDX_workflow_trigger_seat_workflowId_nodeId_seatIndex" ON public.workflow_trigger_seat USING btree ("workflowId", "nodeId", "seatIndex") |
| PK_d88e285cade3e5f2bb626a071f1 | CREATE UNIQUE INDEX "PK_d88e285cade3e5f2bb626a071f1" ON public.workflow_trigger_seat USING btree (id) |
## Relations
```mermaid
erDiagram
"public.workflow_trigger_seat" {
varchar_16_ actualState
varchar_36_ actualVersionId
timestamp_3__with_time_zone createdAt
varchar_255_ desiredHolderId
varchar_16_ desiredState
varchar_36_ desiredVersionId
varchar_255_ holderId
bigint id
text lastError
integer leaseEpoch
timestamp_3__with_time_zone leaseExpiresAt
varchar_36_ nodeId
integer seatIndex
timestamp_3__with_time_zone updatedAt
varchar_36_ workflowId
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

25
docs/generated/sqlite-schema/README.md generated
View File

@ -115,6 +115,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [test_case_execution](test_case_execution.md) | 16 | | table |
| [test_run](test_run.md) | 16 | | table |
| [token_exchange_jti](token_exchange_jti.md) | 3 | | table |
| [trigger_runner](trigger_runner.md) | 4 | | table |
| [trusted_key](trusted_key.md) | 4 | | table |
| [trusted_key_source](trusted_key_source.md) | 8 | | table |
| [user](user.md) | 15 | | table |
@ -137,6 +138,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [workflow_review_request_reviewers](workflow_review_request_reviewers.md) | 2 | | table |
| [workflow_review_request_workflow](workflow_review_request_workflow.md) | 5 | | table |
| [workflow_statistics](workflow_statistics.md) | 7 | | table |
| [workflow_trigger_seat](workflow_trigger_seat.md) | 15 | | table |
| [workflows_tags](workflows_tags.md) | 2 | | table |
## Relations
@ -1402,6 +1404,12 @@ erDiagram
datetime_3_ expiresAt
varchar_255_ jti PK
}
"trigger_runner" {
datetime_3_ createdAt
datetime_3_ lastHeartbeatAt
varchar_255_ runnerId PK
datetime_3_ updatedAt
}
"trusted_key" {
datetime_3_ createdAt
TEXT data
@ -1615,6 +1623,23 @@ erDiagram
VARCHAR_36_ workflowId
VARCHAR_128_ workflowName
}
"workflow_trigger_seat" {
varchar_16_ actualState
varchar_36_ actualVersionId
datetime_3_ createdAt
varchar_255_ desiredHolderId
varchar_16_ desiredState
varchar_36_ desiredVersionId
varchar_255_ holderId
INTEGER id
TEXT lastError
INTEGER leaseEpoch
datetime_3_ leaseExpiresAt
varchar_36_ nodeId
INTEGER seatIndex
datetime_3_ updatedAt
varchar_36_ workflowId
}
"workflows_tags" {
INTEGER tagId PK
varchar_36_ workflowId PK

52
docs/generated/sqlite-schema/trigger_runner.md generated Normal file
View File

@ -0,0 +1,52 @@
# trigger_runner
## Description
<details>
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "trigger_runner" ("runnerId" varchar(255) PRIMARY KEY NOT NULL, "lastHeartbeatAt" datetime(3) NOT NULL, "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')))
```
</details>
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| lastHeartbeatAt | datetime(3) | | false | | | |
| runnerId | varchar(255) | | false | | | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| runnerId | PRIMARY KEY | PRIMARY KEY (runnerId) |
| sqlite_autoindex_trigger_runner_1 | PRIMARY KEY | PRIMARY KEY (runnerId) |
## Indexes
| Name | Definition |
| ---- | ---------- |
| sqlite_autoindex_trigger_runner_1 | PRIMARY KEY (runnerId) |
## Relations
```mermaid
erDiagram
"trigger_runner" {
datetime_3_ createdAt
datetime_3_ lastHeartbeatAt
varchar_255_ runnerId PK
datetime_3_ updatedAt
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

76
docs/generated/sqlite-schema/workflow_trigger_seat.md generated Normal file
View File

@ -0,0 +1,76 @@
# workflow_trigger_seat
## Description
<details>
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "workflow_trigger_seat" ("id" integer PRIMARY KEY NOT NULL, "workflowId" varchar(36) NOT NULL, "nodeId" varchar(36) NOT NULL, "seatIndex" integer NOT NULL, "desiredState" varchar(16) NOT NULL DEFAULT ('active'), "desiredVersionId" varchar(36) NOT NULL, "holderId" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "desiredHolderId" varchar(255), "actualState" varchar(16), "actualVersionId" varchar(36), "lastError" 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_trigger_seat_desiredState" CHECK ("desiredState" IN ('active', 'inactive')), CONSTRAINT "CHK_workflow_trigger_seat_actualState" CHECK ("actualState" IN ('registered', 'closed', 'error')))
```
</details>
## Columns
| Name | Type | Default | Nullable | Children | Parents | Comment |
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
| actualState | varchar(16) | | true | | | |
| actualVersionId | varchar(36) | | true | | | |
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| desiredHolderId | varchar(255) | | true | | | |
| desiredState | varchar(16) | 'active' | false | | | |
| desiredVersionId | varchar(36) | | false | | | |
| holderId | varchar(255) | | true | | | |
| id | INTEGER | | false | | | |
| lastError | TEXT | | true | | | |
| leaseEpoch | INTEGER | 0 | false | | | |
| leaseExpiresAt | datetime(3) | | true | | | |
| nodeId | varchar(36) | | false | | | |
| seatIndex | INTEGER | | false | | | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
| workflowId | varchar(36) | | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| - | CHECK | CHECK ("desiredState" IN ('active', 'inactive')) |
| - | CHECK | CHECK ("actualState" IN ('registered', 'closed', 'error')) |
| id | PRIMARY KEY | PRIMARY KEY (id) |
## Indexes
| Name | Definition |
| ---- | ---------- |
| IDX_workflow_trigger_seat_desiredState | CREATE INDEX "IDX_workflow_trigger_seat_desiredState" ON "workflow_trigger_seat" ("desiredState") |
| IDX_workflow_trigger_seat_workflowId_nodeId_seatIndex | CREATE UNIQUE INDEX "IDX_workflow_trigger_seat_workflowId_nodeId_seatIndex" ON "workflow_trigger_seat" ("workflowId", "nodeId", "seatIndex") |
## Relations
```mermaid
erDiagram
"workflow_trigger_seat" {
varchar_16_ actualState
varchar_36_ actualVersionId
datetime_3_ createdAt
varchar_255_ desiredHolderId
varchar_16_ desiredState
varchar_36_ desiredVersionId
varchar_255_ holderId
INTEGER id
TEXT lastError
INTEGER leaseEpoch
datetime_3_ leaseExpiresAt
varchar_36_ nodeId
INTEGER seatIndex
datetime_3_ updatedAt
varchar_36_ workflowId
}
```
---
> Generated by [tbls](https://github.com/k1LoW/tbls)

18
packages/@n8n/db/src/entities/index.ts
View File

@ -55,6 +55,7 @@ import { SharedWorkflow } from './shared-workflow';
import { TagEntity } from './tag-entity';
import { TestCaseExecution } from './test-case-execution.ee';
import { TestRun } from './test-run.ee';
import { TriggerRunner } from './trigger-runner';
import { User } from './user';
import { Variables } from './variables';
import { WebhookEntity } from './webhook-entity';
@ -88,8 +89,23 @@ import {
} from './workflow-review-request.ee';
import { WorkflowStatistics } from './workflow-statistics';
import { WorkflowTagMapping } from './workflow-tag-mapping';
import {
WorkflowTriggerSeat,
TriggerSeatDesiredStateList,
type TriggerSeatDesiredState,
TriggerSeatActualStateList,
type TriggerSeatActualState,
type TriggerSeatFence,
} from './workflow-trigger-seat';
export {
TriggerRunner,
WorkflowTriggerSeat,
TriggerSeatDesiredStateList,
type TriggerSeatDesiredState,
TriggerSeatActualStateList,
type TriggerSeatActualState,
type TriggerSeatFence,
AgentEvalDataset,
type AgentEvalColumnMapping,
AgentEvalRun,
@ -235,6 +251,8 @@ export const entities = {
Role,
ScheduledJob,
ScheduledTask,
TriggerRunner,
WorkflowTriggerSeat,
ProjectSecretsProviderAccess,
SecretsProviderConnection,
};

18
packages/@n8n/db/src/entities/trigger-runner.ts Normal file
View File

@ -0,0 +1,18 @@
import { Entity, PrimaryColumn } from '@n8n/typeorm';
import { DateTimeColumn, WithTimestamps } from './abstract-entity';
/**
* Liveness registry of instances eligible to hold trigger seats. A runner
* upserts its row on every reconcile tick; a runner whose heartbeat is older
* than the liveness TTL is treated as gone when computing seat assignments.
*/
@Entity({ name: 'trigger_runner' })
export class TriggerRunner extends WithTimestamps {
/** The instance's hostId. */
@PrimaryColumn({ type: 'varchar', length: 255 })
runnerId: string;
@DateTimeColumn()
lastHeartbeatAt: Date;
}

102
packages/@n8n/db/src/entities/workflow-trigger-seat.ts Normal file
View File

@ -0,0 +1,102 @@
import { Column, Entity, Generated, Index, PrimaryColumn } from '@n8n/typeorm';
import { DateTimeColumn, WithTimestamps, dbType } from './abstract-entity';
import { idStringifier } from '../utils/transformers';
export const TriggerSeatDesiredStateList = ['active', 'inactive'] as const;
export type TriggerSeatDesiredState = (typeof TriggerSeatDesiredStateList)[number];
export const TriggerSeatActualStateList = ['registered', 'closed', 'error'] as const;
export type TriggerSeatActualState = (typeof TriggerSeatActualStateList)[number];
/**
* Fencing token a seat holder presents when creating an execution for a
* trigger emission. Checked in the same transaction as the execution insert:
* the seat must still be held by this holder at this epoch, and its desired
* version must match the version the trigger was registered for.
*/
export type TriggerSeatFence = {
seatId: string;
holderId: string;
leaseEpoch: number;
versionId: string;
};
/**
* One leasable replica slot ("seat") of an in-memory trigger node.
*
* A trigger with replication N gets seats 0..N-1. A runner (main instance)
* may hold at most one seat per trigger node, so the number of concurrently
* running replicas is min(N, live runners). The row carries both the desired
* state written by workflow publication (which version should run, whether
* the trigger should run at all) and the lease of the runner currently
* serving the seat.
*
* Every lease write is guarded on `holderId` + `leaseEpoch`; zero rows
* affected means the seat is no longer ours, which is benign. The epoch, not
* the wall-clock expiry, is what fences a stale holder's effects: an
* expired-but-unreclaimed lease may still commit, only a newer claim revokes.
*/
@Entity({ name: 'workflow_trigger_seat' })
@Index(['workflowId', 'nodeId', 'seatIndex'], { unique: true })
@Index(['desiredState'])
export class WorkflowTriggerSeat extends WithTimestamps {
@Generated()
@PrimaryColumn({
type: dbType === 'sqlite' ? 'integer' : 'bigint',
transformer: idStringifier,
})
id: string;
@Column({ type: 'varchar', length: 36 })
workflowId: string;
@Column({ type: 'varchar', length: 36 })
nodeId: string;
/** Which of the trigger's N replica slots this row is. */
@Column({ type: 'int' })
seatIndex: number;
/**
* Whether a runner should be serving this seat. Publication flips this
* instead of deleting rows, so a holder observes the transition and tears
* down cleanly before the row is eventually pruned.
*/
@Column({ type: 'varchar', length: 16, default: 'active' })
desiredState: TriggerSeatDesiredState;
/** The published version the seat's trigger should run. */
@Column({ type: 'varchar', length: 36 })
desiredVersionId: string;
/** Runner currently holding the seat's lease; `null` when vacant. */
@Column({ type: 'varchar', length: 255, nullable: true })
holderId: string | null;
@DateTimeColumn({ nullable: true })
leaseExpiresAt: Date | null;
/** Fencing token, bumped on every claim. */
@Column({ type: 'int', default: 0 })
leaseEpoch: number;
/**
* Set by an underloaded runner to politely request the seat. The current
* holder observes it on its next reconcile tick and releases; the seat is
* never seized from a live holder.
*/
@Column({ type: 'varchar', length: 255, nullable: true })
desiredHolderId: string | null;
/** What the holder last reported doing with the seat. */
@Column({ type: 'varchar', length: 16, nullable: true })
actualState: TriggerSeatActualState | null;
/** The version the holder's registered trigger is actually running. */
@Column({ type: 'varchar', length: 36, nullable: true })
actualVersionId: string | null;
@Column({ type: 'text', nullable: true })
lastError: string | null;
}

79
packages/@n8n/db/src/migrations/common/1787300000000-CreateTriggerSeatTables.ts Normal file
View File

@ -0,0 +1,79 @@
import type { MigrationContext, ReversibleMigration } from '../migration-types';
export class CreateTriggerSeatTables1787300000000 implements ReversibleMigration {
async up(context: MigrationContext) {
await this.createWorkflowTriggerSeatTable(context);
await this.createTriggerRunnerTable(context);
}
async down({ schemaBuilder: { dropTable } }: MigrationContext) {
await dropTable('trigger_runner');
await dropTable('workflow_trigger_seat');
}
private async createWorkflowTriggerSeatTable(context: MigrationContext) {
const {
schemaBuilder: { createTable, createIndex, column },
isSqlite,
} = context;
const idColumn = isSqlite
? column('id').int.primary.autoGenerate2
: column('id').bigint.primary.autoGenerate2;
await createTable('workflow_trigger_seat').withColumns(
idColumn,
column('workflowId').varchar(36).notNull,
column('nodeId').varchar(36).notNull,
column('seatIndex').int.notNull.comment(
'Which of the trigger node’s N replica slots this row is.',
),
column('desiredState')
.varchar(16)
.notNull.default("'active'")
.withEnumCheck(['active', 'inactive'])
.comment(
'Whether a runner should serve this seat; publication flips it instead of deleting rows so holders tear down cleanly.',
),
column('desiredVersionId')
.varchar(36)
.notNull.comment('The published version the seat’s trigger should run.'),
column('holderId')
.varchar(255)
.comment('Runner currently holding the lease; NULL when vacant.'),
column('leaseExpiresAt')
.timestampTimezone()
.comment('When the current lease may be reclaimed by another runner.'),
column('leaseEpoch')
.int.notNull.default(0)
.comment(
'Fencing token bumped on every claim; execution inserts are guarded on it so a stale holder’s emissions never land.',
),
column('desiredHolderId')
.varchar(255)
.comment('Handoff request: an underloaded runner asking the holder to release.'),
column('actualState')
.varchar(16)
.withEnumCheck(['registered', 'closed', 'error'])
.comment('What the holder last reported doing with the seat.'),
column('actualVersionId')
.varchar(36)
.comment('The version the holder’s registered trigger is actually running.'),
column('lastError').text,
).withTimestamps;
await createIndex('workflow_trigger_seat', ['workflowId', 'nodeId', 'seatIndex'], true);
await createIndex('workflow_trigger_seat', ['desiredState']);
}
private async createTriggerRunnerTable({
schemaBuilder: { createTable, column },
}: MigrationContext) {
await createTable('trigger_runner').withColumns(
column('runnerId').varchar(255).primary,
column('lastHeartbeatAt')
.timestampTimezone()
.notNull.comment('Upserted every reconcile tick; older than TTL means the runner is gone.'),
).withTimestamps;
}
}

2
packages/@n8n/db/src/repositories/index.ts
View File

@ -61,6 +61,8 @@ export {
type PollerFailureState,
type PollerFullState,
} from './poller-state.repository';
export { TriggerRunnerRepository } from './trigger-runner.repository';
export { WorkflowTriggerSeatRepository } from './workflow-trigger-seat.repository';
export { ProcessedDataRepository } from './processed-data.repository';
export { SettingsRepository } from './settings.repository';
export { TagRepository } from './tag.repository';

50
packages/@n8n/db/src/repositories/trigger-runner.repository.ts Normal file
View File

@ -0,0 +1,50 @@
import { DatabaseConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import { DataSource } from '@n8n/typeorm';
import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity';
import { TriggerRunner } from '../entities';
import { BaseRepository } from './base-repository';
import type { OperationContext } from '../services/transaction';
import { TransactionRunner } from '../services/transaction';
import { dbNowLiteral, dbNowPlusMsLiteral } from '../utils/dialect-time';
/** Liveness registry of instances eligible to hold trigger seats. */
@Service()
export class TriggerRunnerRepository extends BaseRepository<TriggerRunner> {
private readonly isPostgres: boolean;
constructor(
dataSource: DataSource,
transactionRunner: TransactionRunner,
config: DatabaseConfig,
) {
super(TriggerRunner, dataSource.manager, transactionRunner);
this.isPostgres = config.type === 'postgresdb';
}
async heartbeat(runnerId: string, ctx: OperationContext = {}): Promise<void> {
const now = dbNowLiteral(this.isPostgres);
await this.managerFor(ctx)
.createQueryBuilder()
.insert()
.into(TriggerRunner)
.values({
runnerId,
lastHeartbeatAt: () => now,
} as QueryDeepPartialEntity<TriggerRunner>)
.orUpdate(['lastHeartbeatAt'], ['runnerId'])
.execute();
}
/** Runner ids whose heartbeat is within `ttlMs`, sorted for determinism. */
async findLiveRunnerIds(ttlMs: number, ctx: OperationContext = {}): Promise<string[]> {
const rows: Array<{ runnerId: string }> = await this.managerFor(ctx)
.createQueryBuilder(TriggerRunner, 'runner')
.select('runner.runnerId', 'runnerId')
.where(`runner.lastHeartbeatAt > ${dbNowPlusMsLiteral(this.isPostgres, -ttlMs)}`)
.orderBy('runner.runnerId', 'ASC')
.getRawMany();
return rows.map((row) => row.runnerId);
}
}

275
packages/@n8n/db/src/repositories/workflow-trigger-seat.repository.ts Normal file
View File

@ -0,0 +1,275 @@
import { DatabaseConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import { DataSource, In, MoreThanOrEqual } from '@n8n/typeorm';
import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity';
import { WorkflowTriggerSeat } from '../entities';
import { BaseRepository } from './base-repository';
import type { TriggerSeatActualState, TriggerSeatFence } from '../entities/workflow-trigger-seat';
import type { OperationContext } from '../services/transaction';
import { TransactionRunner } from '../services/transaction';
import { dbNowLiteral, dbNowPlusMsLiteral } from '../utils/dialect-time';
/**
* Leasable replica slots ("seats") of in-memory trigger nodes.
*
* Every lease write is guarded on `holderId` + `leaseEpoch`; zero rows
* affected means the seat is no longer ours and is benign, never an error.
* The claim is the only write that bumps the epoch, so the epoch is the
* fencing token: effects of a holder whose seat was re-claimed never land.
*/
@Service()
export class WorkflowTriggerSeatRepository extends BaseRepository<WorkflowTriggerSeat> {
private readonly isPostgres: boolean;
constructor(
dataSource: DataSource,
transactionRunner: TransactionRunner,
config: DatabaseConfig,
) {
super(WorkflowTriggerSeat, dataSource.manager, transactionRunner);
this.isPostgres = config.type === 'postgresdb';
}
/**
* Projects a trigger node's desired replication: seats `0..seatCount-1`
* become active at `versionId`, any higher-indexed leftovers become
* inactive. Idempotent; existing leases are left untouched so holders
* observe the change and converge.
*/
async upsertDesiredSeats(
workflowId: string,
nodeId: string,
seatCount: number,
versionId: string,
ctx: OperationContext = {},
): Promise<void> {
const manager = this.managerFor(ctx);
for (let seatIndex = 0; seatIndex < seatCount; seatIndex++) {
const updated = await manager
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
desiredState: 'active',
desiredVersionId: versionId,
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ workflowId, nodeId, seatIndex })
.execute();
if (updated.affected === 0) {
await manager
.createQueryBuilder()
.insert()
.into(WorkflowTriggerSeat)
.values({
workflowId,
nodeId,
seatIndex,
desiredState: 'active',
desiredVersionId: versionId,
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.orIgnore()
.execute();
}
}
await manager
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
desiredState: 'inactive',
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ workflowId, nodeId, seatIndex: MoreThanOrEqual(seatCount) })
.execute();
}
/** Marks every seat of the given nodes inactive, so holders tear down. */
async markSeatsInactive(
workflowId: string,
nodeIds: string[],
ctx: OperationContext = {},
): Promise<void> {
if (nodeIds.length === 0) return;
await this.managerFor(ctx)
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
desiredState: 'inactive',
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ workflowId, nodeId: In(nodeIds) })
.execute();
}
async deleteSeatsForWorkflow(workflowId: string, ctx: OperationContext = {}): Promise<void> {
await this.managerFor(ctx).delete(WorkflowTriggerSeat, { workflowId });
}
/**
* Every seat a reconcile tick cares about: seats that should be running,
* plus inactive seats still held (their holders owe a teardown).
*/
async findAllRelevantSeats(ctx: OperationContext = {}): Promise<WorkflowTriggerSeat[]> {
return await this.managerFor(ctx)
.createQueryBuilder(WorkflowTriggerSeat, 'seat')
.where("seat.desiredState = 'active'")
.orWhere('seat.holderId IS NOT NULL')
.getMany();
}
/**
* Claims a vacant or lease-expired active seat, bumping the fencing epoch.
*
* @returns The new lease epoch, or `null` if the seat was not claimable
* (held with a live lease, inactive, or gone).
*/
async claim(
seatId: string,
runnerId: string,
leaseMs: number,
ctx: OperationContext = {},
): Promise<number | null> {
const manager = this.managerFor(ctx);
const result = await manager
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
holderId: runnerId,
leaseExpiresAt: () => dbNowPlusMsLiteral(this.isPostgres, leaseMs),
leaseEpoch: () => '"leaseEpoch" + 1',
desiredHolderId: null,
lastError: null,
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where('id = :seatId', { seatId })
.andWhere('"desiredState" = \'active\'')
.andWhere(`("holderId" IS NULL OR "leaseExpiresAt" < ${dbNowLiteral(this.isPostgres)})`)
.execute();
if (result.affected !== 1) return null;
const row = await manager.findOne(WorkflowTriggerSeat, {
select: ['leaseEpoch'],
where: { id: seatId, holderId: runnerId },
});
return row?.leaseEpoch ?? null;
}
/** Extends the lease. `false` means the seat is no longer ours. */
async renew(
seatId: string,
runnerId: string,
leaseEpoch: number,
leaseMs: number,
ctx: OperationContext = {},
): Promise<boolean> {
const result = await this.managerFor(ctx)
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
leaseExpiresAt: () => dbNowPlusMsLiteral(this.isPostgres, leaseMs),
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ id: seatId, holderId: runnerId, leaseEpoch })
.execute();
return result.affected === 1;
}
/** Vacates the seat without bumping the epoch. `false` means it wasn't ours. */
async release(
seatId: string,
runnerId: string,
leaseEpoch: number,
ctx: OperationContext = {},
): Promise<boolean> {
const result = await this.managerFor(ctx)
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
holderId: null,
leaseExpiresAt: null,
desiredHolderId: null,
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ id: seatId, holderId: runnerId, leaseEpoch })
.execute();
return result.affected === 1;
}
/** Records what the holder is actually doing with the seat. */
async reportActual(
seatId: string,
runnerId: string,
leaseEpoch: number,
actual: { state: TriggerSeatActualState; versionId?: string | null; error?: string | null },
ctx: OperationContext = {},
): Promise<boolean> {
const result = await this.managerFor(ctx)
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
actualState: actual.state,
actualVersionId: actual.versionId ?? null,
lastError: actual.error ?? null,
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ id: seatId, holderId: runnerId, leaseEpoch })
.execute();
return result.affected === 1;
}
/**
* Politely asks the current holder to release the seat. Set only when no
* other handoff is pending; the holder honors it on its next tick.
*/
async requestHandoff(
seatId: string,
toRunnerId: string,
ctx: OperationContext = {},
): Promise<boolean> {
const result = await this.managerFor(ctx)
.createQueryBuilder()
.update(WorkflowTriggerSeat)
.set({
desiredHolderId: toRunnerId,
updatedAt: () => dbNowLiteral(this.isPostgres),
} as QueryDeepPartialEntity<WorkflowTriggerSeat>)
.where({ id: seatId })
.andWhere('"desiredHolderId" IS NULL')
.andWhere('"holderId" IS NOT NULL')
.andWhere('"holderId" != :toRunnerId', { toRunnerId })
.execute();
return result.affected === 1;
}
/**
* The fence: asserts the emitting holder still owns the seat at this epoch
* and that the seat still wants the version the trigger was registered for.
* Run inside the same transaction as the execution insert. On Postgres the
* row is locked FOR SHARE so a concurrent claim can't slip between the
* check and the commit; on SQLite the single-writer model makes the plain
* read equivalent.
*
* Deliberately does not check `leaseExpiresAt`: an expired-but-unreclaimed
* lease still commits. Correctness comes from epoch monotonicity, not clocks.
*/
async assertSeatHeld(fence: TriggerSeatFence, ctx: OperationContext): Promise<boolean> {
const manager = this.managerFor(ctx);
const row = await manager.findOne(WorkflowTriggerSeat, {
select: ['id'],
where: {
id: fence.seatId,
holderId: fence.holderId,
leaseEpoch: fence.leaseEpoch,
desiredVersionId: fence.versionId,
desiredState: 'active',
},
...(this.isPostgres ? { lock: { mode: 'pessimistic_read' as const } } : {}),
});
return row !== null;
}
}

241
packages/cli/test/integration/trigger-seats/seat-repositories.test.ts Normal file
View File

@ -0,0 +1,241 @@
import { testDb } from '@n8n/backend-test-utils';
import type { WorkflowTriggerSeat } from '@n8n/db';
import { TriggerRunnerRepository, WorkflowTriggerSeatRepository } from '@n8n/db';
import { Container } from '@n8n/di';
// Exercises the seat lease semantics directly against the database: claims bump
// the fencing epoch, every other write is guarded on holder+epoch, and the
// fence assertion only passes for the live holder at the desired version.
describe('trigger seat repositories', () => {
let seatRepository: WorkflowTriggerSeatRepository;
let runnerRepository: TriggerRunnerRepository;
const LEASE_MS = 30_000;
const EXPIRED_LEASE_MS = -1_000;
beforeAll(async () => {
await testDb.init();
seatRepository = Container.get(WorkflowTriggerSeatRepository);
runnerRepository = Container.get(TriggerRunnerRepository);
});
beforeEach(async () => {
await testDb.truncate(['WorkflowTriggerSeat', 'TriggerRunner']);
});
afterAll(async () => {
await testDb.terminate();
});
async function createSeat(
overrides: Partial<WorkflowTriggerSeat> = {},
): Promise<WorkflowTriggerSeat> {
await seatRepository.upsertDesiredSeats(
overrides.workflowId ?? 'wf-1',
overrides.nodeId ?? 'node-1',
1,
overrides.desiredVersionId ?? 'v1',
);
const seat = await seatRepository.findOneOrFail({
where: { workflowId: overrides.workflowId ?? 'wf-1', nodeId: overrides.nodeId ?? 'node-1' },
});
return seat;
}
describe('claim', () => {
it('claims a vacant seat and bumps the epoch', async () => {
const seat = await createSeat();
expect(seat.leaseEpoch).toBe(0);
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(epoch).toBe(1);
const after = await seatRepository.findOneOrFail({ where: { id: seat.id } });
expect(after.holderId).toBe('runner-a');
expect(after.leaseExpiresAt).not.toBeNull();
});
it('refuses a seat held with a live lease', async () => {
const seat = await createSeat();
await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
const epoch = await seatRepository.claim(seat.id, 'runner-b', LEASE_MS);
expect(epoch).toBeNull();
});
it('reclaims a seat whose lease expired, bumping the epoch again', async () => {
const seat = await createSeat();
await seatRepository.claim(seat.id, 'runner-a', EXPIRED_LEASE_MS);
const epoch = await seatRepository.claim(seat.id, 'runner-b', LEASE_MS);
expect(epoch).toBe(2);
const after = await seatRepository.findOneOrFail({ where: { id: seat.id } });
expect(after.holderId).toBe('runner-b');
});
it('refuses an inactive seat', async () => {
const seat = await createSeat();
await seatRepository.markSeatsInactive(seat.workflowId, [seat.nodeId]);
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(epoch).toBeNull();
});
});
describe('renew / release / reportActual', () => {
it('renews only for the current holder and epoch', async () => {
const seat = await createSeat();
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(await seatRepository.renew(seat.id, 'runner-a', epoch!, LEASE_MS)).toBe(true);
expect(await seatRepository.renew(seat.id, 'runner-a', epoch! - 1, LEASE_MS)).toBe(false);
expect(await seatRepository.renew(seat.id, 'runner-b', epoch!, LEASE_MS)).toBe(false);
});
it('release vacates without bumping the epoch; the next claim bumps it', async () => {
const seat = await createSeat();
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(await seatRepository.release(seat.id, 'runner-a', epoch!)).toBe(true);
const vacated = await seatRepository.findOneOrFail({ where: { id: seat.id } });
expect(vacated.holderId).toBeNull();
expect(vacated.leaseEpoch).toBe(1);
expect(await seatRepository.claim(seat.id, 'runner-b', LEASE_MS)).toBe(2);
});
it('reportActual is guarded on holder and epoch', async () => {
const seat = await createSeat();
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(
await seatRepository.reportActual(seat.id, 'runner-a', epoch!, {
state: 'registered',
versionId: 'v1',
}),
).toBe(true);
expect(
await seatRepository.reportActual(seat.id, 'runner-b', epoch!, { state: 'closed' }),
).toBe(false);
const after = await seatRepository.findOneOrFail({ where: { id: seat.id } });
expect(after.actualState).toBe('registered');
expect(after.actualVersionId).toBe('v1');
});
});
describe('assertSeatHeld (the fence)', () => {
it('passes for the live holder at the desired version', async () => {
const seat = await createSeat({ desiredVersionId: 'v1' });
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
const held = await seatRepository.assertSeatHeld(
{ seatId: seat.id, holderId: 'runner-a', leaseEpoch: epoch!, versionId: 'v1' },
{},
);
expect(held).toBe(true);
});
it('rejects a zombie whose seat was reclaimed', async () => {
const seat = await createSeat();
const staleEpoch = await seatRepository.claim(seat.id, 'runner-a', EXPIRED_LEASE_MS);
await seatRepository.claim(seat.id, 'runner-b', LEASE_MS);
const held = await seatRepository.assertSeatHeld(
{ seatId: seat.id, holderId: 'runner-a', leaseEpoch: staleEpoch!, versionId: 'v1' },
{},
);
expect(held).toBe(false);
});
it('still passes for a holder whose lease expired but was not reclaimed', async () => {
// Epoch-not-expiry: only a newer claim revokes, never the clock alone.
const seat = await createSeat();
const epoch = await seatRepository.claim(seat.id, 'runner-a', EXPIRED_LEASE_MS);
const held = await seatRepository.assertSeatHeld(
{ seatId: seat.id, holderId: 'runner-a', leaseEpoch: epoch!, versionId: 'v1' },
{},
);
expect(held).toBe(true);
});
it('rejects when the desired version moved on', async () => {
const seat = await createSeat({ desiredVersionId: 'v1' });
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
await seatRepository.upsertDesiredSeats(seat.workflowId, seat.nodeId, 1, 'v2');
const held = await seatRepository.assertSeatHeld(
{ seatId: seat.id, holderId: 'runner-a', leaseEpoch: epoch!, versionId: 'v1' },
{},
);
expect(held).toBe(false);
});
it('rejects when the seat went inactive', async () => {
const seat = await createSeat();
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
await seatRepository.markSeatsInactive(seat.workflowId, [seat.nodeId]);
const held = await seatRepository.assertSeatHeld(
{ seatId: seat.id, holderId: 'runner-a', leaseEpoch: epoch!, versionId: 'v1' },
{},
);
expect(held).toBe(false);
});
});
describe('upsertDesiredSeats', () => {
it('creates N seats and marks shrunk leftovers inactive', async () => {
await seatRepository.upsertDesiredSeats('wf-1', 'node-1', 3, 'v1');
expect(await seatRepository.count({ where: { desiredState: 'active' } })).toBe(3);
await seatRepository.upsertDesiredSeats('wf-1', 'node-1', 1, 'v1');
const seats = await seatRepository.find({ order: { seatIndex: 'ASC' } });
expect(seats.map((seat) => seat.desiredState)).toEqual(['active', 'inactive', 'inactive']);
});
it('retargets the version without touching an existing lease', async () => {
const seat = await createSeat({ desiredVersionId: 'v1' });
const epoch = await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
await seatRepository.upsertDesiredSeats(seat.workflowId, seat.nodeId, 1, 'v2');
const after = await seatRepository.findOneOrFail({ where: { id: seat.id } });
expect(after.desiredVersionId).toBe('v2');
expect(after.holderId).toBe('runner-a');
expect(after.leaseEpoch).toBe(epoch);
});
});
describe('requestHandoff', () => {
it('sets the request once, and never toward the current holder', async () => {
const seat = await createSeat();
await seatRepository.claim(seat.id, 'runner-a', LEASE_MS);
expect(await seatRepository.requestHandoff(seat.id, 'runner-a')).toBe(false);
expect(await seatRepository.requestHandoff(seat.id, 'runner-b')).toBe(true);
expect(await seatRepository.requestHandoff(seat.id, 'runner-c')).toBe(false);
});
});
describe('trigger runner registry', () => {
it('heartbeat upserts and liveness filters by TTL', async () => {
await runnerRepository.heartbeat('runner-a');
await runnerRepository.heartbeat('runner-b');
await runnerRepository.heartbeat('runner-a');
expect(await runnerRepository.findLiveRunnerIds(60_000)).toEqual(['runner-a', 'runner-b']);
expect(await runnerRepository.findLiveRunnerIds(-1_000)).toEqual([]);
});
});
});