github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-16 21:19:43 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Add a timeout column to the scheduler tables (no-changelog) (#40510)

Browse Source
This commit is contained in:
Lorent Lempereur 2026-10-12 08:08:45 +00:00 committed by GitHub
parent 30d883e866
commit 1f27b98366
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 328 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -123,8 +123,8 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
| [public.role_mapping_rule](public.role_mapping_rule.md) | 7 | | BASE TABLE |
| [public.role_mapping_rule_project](public.role_mapping_rule_project.md) | 2 | | BASE TABLE |
| [public.role_scope](public.role_scope.md) | 2 | | BASE TABLE |
| [public.scheduled_job](public.scheduled_job.md) | 24 | | BASE TABLE |
| [public.scheduled_task](public.scheduled_task.md) | 18 | | BASE TABLE |
| [public.scheduled_job](public.scheduled_job.md) | 25 | | BASE TABLE |
| [public.scheduled_task](public.scheduled_task.md) | 19 | | BASE TABLE |
| [public.scope](public.scope.md) | 3 | | BASE TABLE |
| [public.secrets_provider_connection](public.secrets_provider_connection.md) | 7 | | BASE TABLE |
| [public.self_healing_result](public.self_healing_result.md) | 15 | | BASE TABLE |
@ -1576,6 +1576,7 @@ erDiagram
integer recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
integer timeoutSeconds
varchar_64_ timezone
timestamp_3__with_time_zone updatedAt
}
@ -1598,6 +1599,7 @@ erDiagram
timestamp_3__with_time_zone startedAt
varchar_16_ status
varchar_128_ taskType
integer timeoutSeconds
}
"public.scope" {
text description

5
docs/generated/postgres-schema/public.scheduled_job.md generated
View File

@ -26,6 +26,7 @@
| recurrenceSize | integer | | true | | | The N in a recurring_cron schedule's every-N-periods filter, e.g. 3 for every 3 weeks; at least 2. Set only when kind is 'recurring_cron'. |
| recurrenceUnit | varchar(16) | | true | | | Calendar period counted by a recurring_cron schedule's every-N-periods filter (hours, days, weeks, months). Set only when kind is 'recurring_cron'. |
| taskType | varchar(128) | | false | | | Selects which registered handler runs the task. |
| timeoutSeconds | integer | 300 | false | | | How long, in seconds, one attempt of an occurrence may run before the executor stops it. |
| timezone | varchar(64) | | true | | | IANA timezone the cron expression is evaluated in; NULL uses the instance default. |
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
@ -43,6 +44,7 @@
| CHK_scheduled_job_recurrence_size | CHECK | CHECK (("recurrenceSize" >= 2)) |
| CHK_scheduled_job_recurrence_unit | CHECK | CHECK ((("recurrenceUnit")::text = ANY ((ARRAY['hours'::character varying, 'days'::character varying, 'weeks'::character varying, 'months'::character varying])::text[]))) |
| CHK_scheduled_job_recurring_cron | CHECK | CHECK ((((kind)::text <> 'recurring_cron'::text) OR (("cronExpression" IS NOT NULL) AND ("recurrenceUnit" IS NOT NULL) AND ("recurrenceSize" IS NOT NULL)))) |
| CHK_scheduled_job_timeoutSeconds | CHECK | CHECK ((("timeoutSeconds" > 0) AND ("timeoutSeconds" <= 2147483))) NOT VALID |
| PK_893185383f029ca8d57bb781fa8 | PRIMARY KEY | PRIMARY KEY (id) |
| scheduled_job_createdAt_not_null | n | NOT NULL "createdAt" |
| scheduled_job_enabled_not_null | n | NOT NULL enabled |
@ -56,6 +58,7 @@
| scheduled_job_ownerType_not_null | n | NOT NULL "ownerType" |
| scheduled_job_payload_not_null | n | NOT NULL payload |
| scheduled_job_taskType_not_null | n | NOT NULL "taskType" |
| scheduled_job_timeoutSeconds_not_null | n | NOT NULL "timeoutSeconds" |
| scheduled_job_updatedAt_not_null | n | NOT NULL "updatedAt" |
## Indexes
@ -98,6 +101,7 @@ erDiagram
integer recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
integer timeoutSeconds
varchar_64_ timezone
timestamp_3__with_time_zone updatedAt
}
@ -120,6 +124,7 @@ erDiagram
timestamp_3__with_time_zone startedAt
varchar_16_ status
varchar_128_ taskType
integer timeoutSeconds
}
```

5
docs/generated/postgres-schema/public.scheduled_task.md generated
View File

@ -22,6 +22,7 @@
| startedAt | timestamp(3) with time zone | | true | | | When the current attempt started running. |
| status | varchar(16) | 'pending'::character varying | false | | | Lifecycle state; drives which occurrences the claim and reaper scans consider. |
| taskType | varchar(128) | | false | | | What kind of work to run, copied from the job so a run is self-contained (no join to execute it). Also lets a run exist without a parent job in future. |
| timeoutSeconds | integer | 300 | false | | | How long, in seconds, one attempt of an occurrence may run before the executor stops it. |
## Constraints
@ -29,6 +30,7 @@
| ---- | ---- | ---------- |
| CHK_scheduled_task_running_lease | CHECK | CHECK ((((status)::text <> 'running'::text) OR ("leaseExpiresAt" IS NOT NULL))) |
| CHK_scheduled_task_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['pending'::character varying, 'running'::character varying, 'succeeded'::character varying, 'failed'::character varying, 'missed'::character varying, 'cancelled'::character varying])::text[]))) |
| CHK_scheduled_task_timeoutSeconds | CHECK | CHECK ((("timeoutSeconds" > 0) AND ("timeoutSeconds" <= 2147483))) NOT VALID |
| FK_scheduled_task_jobId | FOREIGN KEY | FOREIGN KEY ("jobId") REFERENCES scheduled_job(id) ON DELETE CASCADE |
| PK_d690af24e57e30594c1948af1e6 | PRIMARY KEY | PRIMARY KEY (id) |
| scheduled_task_attempts_not_null | n | NOT NULL attempts |
@ -42,6 +44,7 @@
| scheduled_task_scheduledFor_not_null | n | NOT NULL "scheduledFor" |
| scheduled_task_status_not_null | n | NOT NULL status |
| scheduled_task_taskType_not_null | n | NOT NULL "taskType" |
| scheduled_task_timeoutSeconds_not_null | n | NOT NULL "timeoutSeconds" |
## Indexes
@ -80,6 +83,7 @@ erDiagram
timestamp_3__with_time_zone startedAt
varchar_16_ status
varchar_128_ taskType
integer timeoutSeconds
}
"public.scheduled_job" {
integer concurrencyLimit
@ -104,6 +108,7 @@ erDiagram
integer recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
integer timeoutSeconds
varchar_64_ timezone
timestamp_3__with_time_zone updatedAt
}

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

@ -123,8 +123,8 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
| [role_mapping_rule](role_mapping_rule.md) | 7 | | table |
| [role_mapping_rule_project](role_mapping_rule_project.md) | 2 | | table |
| [role_scope](role_scope.md) | 2 | | table |
| [scheduled_job](scheduled_job.md) | 24 | | table |
| [scheduled_task](scheduled_task.md) | 18 | | table |
| [scheduled_job](scheduled_job.md) | 25 | | table |
| [scheduled_task](scheduled_task.md) | 19 | | table |
| [scope](scope.md) | 3 | | table |
| [secrets_provider_connection](secrets_provider_connection.md) | 7 | | table |
| [self_healing_result](self_healing_result.md) | 15 | | table |
@ -1562,6 +1562,7 @@ erDiagram
INT recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
INT timeoutSeconds
varchar_64_ timezone
datetime_3_ updatedAt
}
@ -1584,6 +1585,7 @@ erDiagram
datetime_3_ startedAt
varchar_16_ status
varchar_128_ taskType
INT timeoutSeconds
}
"scope" {
TEXT description

6
docs/generated/sqlite-schema/scheduled_job.md generated
View File

@ -6,7 +6,7 @@
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(255) NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "kind" varchar(16) NOT NULL, "cronExpression" varchar(255), "timezone" varchar(64), "intervalSeconds" integer, "fireAt" datetime(3), "enabled" boolean NOT NULL DEFAULT (true), "nextRunAt" datetime(3), "lastFiredAt" datetime(3), "maxAttempts" integer NOT NULL DEFAULT (1), "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')), "recurrenceUnit" varchar(16), "recurrenceSize" int, "misfirePolicy" varchar(16) NOT NULL DEFAULT ('coalesce'), "misfireGraceSeconds" int NOT NULL DEFAULT (60), "ownerType" varchar(32) NOT NULL, "ownerId" varchar(255) NOT NULL, "ownerMemberId" varchar(36), "orphanedAt" datetime(3), "concurrencyLimit" int CONSTRAINT "CHK_scheduled_job_concurrencyLimit" CHECK ("concurrencyLimit" IS NULL OR ("concurrencyLimit" >= 1 AND "concurrencyLimit" <= 2147483647 AND CAST("concurrencyLimit" AS INTEGER) = "concurrencyLimit")), CONSTRAINT "CHK_scheduled_job_misfireGraceSeconds" CHECK ((((("misfireGraceSeconds" > 0))))), CONSTRAINT "CHK_scheduled_job_recurrence_unit" CHECK (((((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))))))), CONSTRAINT "CHK_scheduled_job_recurrence_size" CHECK (((((("recurrenceSize" >= 2)))))), CONSTRAINT "CHK_scheduled_job_cron_expression" CHECK (((((("kind" <> 'cron' OR "cronExpression" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_interval_seconds" CHECK (((((("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_fire_at" CHECK (((((("kind" <> 'one_off' OR "fireAt" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_kind" CHECK ((((("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron')))))), CONSTRAINT "CHK_scheduled_job_recurring_cron" CHECK ((((("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_misfirePolicy" CHECK (((("misfirePolicy" IN ('coalesce', 'skip', 'coalesce_owner'))))))
CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(255) NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "kind" varchar(16) NOT NULL, "cronExpression" varchar(255), "timezone" varchar(64), "intervalSeconds" integer, "fireAt" datetime(3), "enabled" boolean NOT NULL DEFAULT (true), "nextRunAt" datetime(3), "lastFiredAt" datetime(3), "maxAttempts" integer NOT NULL DEFAULT (1), "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')), "recurrenceUnit" varchar(16), "recurrenceSize" int, "misfirePolicy" varchar(16) NOT NULL DEFAULT ('coalesce'), "misfireGraceSeconds" int NOT NULL DEFAULT (60), "ownerType" varchar(32) NOT NULL, "ownerId" varchar(255) NOT NULL, "ownerMemberId" varchar(36), "orphanedAt" datetime(3), "concurrencyLimit" int CONSTRAINT "CHK_scheduled_job_concurrencyLimit" CHECK ("concurrencyLimit" IS NULL OR ("concurrencyLimit" >= 1 AND "concurrencyLimit" <= 2147483647 AND CAST("concurrencyLimit" AS INTEGER) = "concurrencyLimit")), "timeoutSeconds" int NOT NULL DEFAULT 300 CONSTRAINT "CHK_scheduled_job_timeoutSeconds" CHECK ("timeoutSeconds" > 0 AND "timeoutSeconds" <= 2147483 AND CAST("timeoutSeconds" AS INTEGER) = "timeoutSeconds"), CONSTRAINT "CHK_scheduled_job_misfireGraceSeconds" CHECK ((((("misfireGraceSeconds" > 0))))), CONSTRAINT "CHK_scheduled_job_recurrence_unit" CHECK (((((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))))))), CONSTRAINT "CHK_scheduled_job_recurrence_size" CHECK (((((("recurrenceSize" >= 2)))))), CONSTRAINT "CHK_scheduled_job_cron_expression" CHECK (((((("kind" <> 'cron' OR "cronExpression" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_interval_seconds" CHECK (((((("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_fire_at" CHECK (((((("kind" <> 'one_off' OR "fireAt" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_kind" CHECK ((((("kind" IN ('cron', 'interval', 'one_off', 'recurring_cron')))))), CONSTRAINT "CHK_scheduled_job_recurring_cron" CHECK ((((("kind" <> 'recurring_cron' OR ("cronExpression" IS NOT NULL AND "recurrenceUnit" IS NOT NULL AND "recurrenceSize" IS NOT NULL)))))), CONSTRAINT "CHK_scheduled_job_misfirePolicy" CHECK (((("misfirePolicy" IN ('coalesce', 'skip', 'coalesce_owner'))))))
```
</details>
@ -37,6 +37,7 @@ CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(
| recurrenceSize | INT | | true | | | |
| recurrenceUnit | varchar(16) | | true | | | |
| taskType | varchar(128) | | false | | | |
| timeoutSeconds | INT | 300 | false | | | |
| timezone | varchar(64) | | true | | | |
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
@ -45,6 +46,7 @@ CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(
| Name | Type | Definition |
| ---- | ---- | ---------- |
| - | CHECK | CHECK ("concurrencyLimit" IS NULL OR ("concurrencyLimit" >= 1 AND "concurrencyLimit" <= 2147483647 AND CAST("concurrencyLimit" AS INTEGER) = "concurrencyLimit")) |
| - | CHECK | CHECK ("timeoutSeconds" > 0 AND "timeoutSeconds" <= 2147483 AND CAST("timeoutSeconds" AS INTEGER) = "timeoutSeconds") |
| - | CHECK | CHECK ((((("misfireGraceSeconds" > 0))))) |
| - | CHECK | CHECK (((((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))))))) |
| - | CHECK | CHECK (((((("recurrenceSize" >= 2)))))) |
@ -95,6 +97,7 @@ erDiagram
INT recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
INT timeoutSeconds
varchar_64_ timezone
datetime_3_ updatedAt
}
@ -117,6 +120,7 @@ erDiagram
datetime_3_ startedAt
varchar_16_ status
varchar_128_ taskType
INT timeoutSeconds
}
```

6
docs/generated/sqlite-schema/scheduled_task.md generated
View File

@ -6,7 +6,7 @@
<summary><strong>Table Definition</strong></summary>
```sql
CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" integer NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "scheduledFor" datetime(3) NOT NULL, "runAt" datetime(3) NOT NULL, "status" varchar(16) NOT NULL DEFAULT ('pending'), "attempts" integer NOT NULL DEFAULT (0), "maxAttempts" integer NOT NULL DEFAULT (1), "claimedBy" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "startedAt" datetime(3), "finishedAt" datetime(3), "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "dispatchedAt" datetime(3), "missedAfter" datetime(3), CONSTRAINT "CHK_scheduled_task_running_lease" CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_task_status" CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')), CONSTRAINT "FK_scheduled_task_jobId" FOREIGN KEY ("jobId") REFERENCES "scheduled_job" ("id") ON DELETE CASCADE)
CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" integer NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "scheduledFor" datetime(3) NOT NULL, "runAt" datetime(3) NOT NULL, "status" varchar(16) NOT NULL DEFAULT ('pending'), "attempts" integer NOT NULL DEFAULT (0), "maxAttempts" integer NOT NULL DEFAULT (1), "claimedBy" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "startedAt" datetime(3), "finishedAt" datetime(3), "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "dispatchedAt" datetime(3), "missedAfter" datetime(3), "timeoutSeconds" int NOT NULL DEFAULT 300 CONSTRAINT "CHK_scheduled_task_timeoutSeconds" CHECK ("timeoutSeconds" > 0 AND "timeoutSeconds" <= 2147483 AND CAST("timeoutSeconds" AS INTEGER) = "timeoutSeconds"), CONSTRAINT "CHK_scheduled_task_running_lease" CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_task_status" CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')), CONSTRAINT "FK_scheduled_task_jobId" FOREIGN KEY ("jobId") REFERENCES "scheduled_job" ("id") ON DELETE CASCADE)
```
</details>
@ -33,11 +33,13 @@ CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" intege
| startedAt | datetime(3) | | true | | | |
| status | varchar(16) | 'pending' | false | | | |
| taskType | varchar(128) | | false | | | |
| timeoutSeconds | INT | 300 | false | | | |
## Constraints
| Name | Type | Definition |
| ---- | ---- | ---------- |
| - | CHECK | CHECK ("timeoutSeconds" > 0 AND "timeoutSeconds" <= 2147483 AND CAST("timeoutSeconds" AS INTEGER) = "timeoutSeconds") |
| - | CHECK | CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL) |
| - | CHECK | CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')) |
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
@ -79,6 +81,7 @@ erDiagram
datetime_3_ startedAt
varchar_16_ status
varchar_128_ taskType
INT timeoutSeconds
}
"scheduled_job" {
INT concurrencyLimit
@ -103,6 +106,7 @@ erDiagram
INT recurrenceSize
varchar_16_ recurrenceUnit
varchar_128_ taskType
INT timeoutSeconds
varchar_64_ timezone
datetime_3_ updatedAt
}

82
packages/@n8n/db/src/migrations/common/1791542683581-AddTimeoutSecondsToScheduler.ts Normal file
View File

@ -0,0 +1,82 @@
import type { MigrationContext, ReversibleMigration } from '../migration-types';
const tables = ['scheduled_job', 'scheduled_task'];
const column = 'timeoutSeconds';
// Pinned rather than read from the config: a migration has to keep meaning what it
// meant when it ran.
const defaultTimeoutSeconds = 300;
// The longest delay a Node.js timer honors, in whole seconds.
const maxTimeoutSeconds = 2147483;
/**
* Adds the timeout of one run to `scheduled_job`, and the copy each occurrence
* takes of it to `scheduled_task`. Both are NOT NULL with a default, so existing
* rows need no backfill.
*/
export class AddTimeoutSecondsToScheduler1791542683581 implements ReversibleMigration {
async up(context: MigrationContext) {
for (const table of tables) {
await this.addTimeoutColumn(context, table);
}
}
/**
* Each CHECK goes before its column: SQLite refuses to drop a column that a
* table-level CHECK names, which a later TypeORM rebuild makes of this one.
* Then a raw `DROP COLUMN`, for the same reason `up` adds it raw.
*/
async down({ queryRunner, runQuery, escape, tablePrefix }: MigrationContext) {
for (const table of tables) {
// Reload the table: a cached copy may predate the raw `ADD COLUMN` in `up`.
await queryRunner.getTable(`${tablePrefix}${table}`);
await queryRunner.dropCheckConstraint(
`${tablePrefix}${table}`,
checkName(tablePrefix, table),
);
await runQuery(
`ALTER TABLE ${escape.tableName(table)} DROP COLUMN ${escape.columnName(column)}`,
);
}
}
/**
* Raw `ADD COLUMN` rather than the schema builder's `addColumns`, so SQLite does
* not rebuild `scheduled_task`, the largest table here, nor `scheduled_job`,
* which `scheduled_task` references with ON DELETE CASCADE.
*/
private async addTimeoutColumn(
{ runQuery, escape, tablePrefix, isPostgres }: MigrationContext,
table: string,
) {
const tableName = escape.tableName(table);
const columnName = escape.columnName(column);
// A timeout of 0 stops every run as soon as it starts. CAST rejects a fractional
// value, which SQLite stores as given. A Postgres int column never holds one.
const integerOnly = isPostgres ? '' : ` AND CAST(${columnName} AS INTEGER) = ${columnName}`;
const check =
`CONSTRAINT "${checkName(tablePrefix, table)}" CHECK (${columnName} > 0 AND ` +
`${columnName} <= ${maxTimeoutSeconds}${integerOnly})`;
const addColumn = `ALTER TABLE ${tableName} ADD COLUMN ${columnName} int NOT NULL DEFAULT ${defaultTimeoutSeconds}`;
if (!isPostgres) {
await runQuery(`${addColumn} ${check}`);
return;
}
// NOT VALID skips the scan that checks the existing rows under the exclusive
// lock. Each of them holds the default, which the CHECK accepts.
await runQuery(addColumn);
await runQuery(`ALTER TABLE ${tableName} ADD ${check} NOT VALID`);
await runQuery(
`COMMENT ON COLUMN ${tableName}.${columnName} IS ` +
"'How long, in seconds, one attempt of an occurrence may run before the executor stops it.'",
);
}
}
/** The name the DSL gives a column CHECK, so `dropCheckConstraint` can find it. */
function checkName(tablePrefix: string, table: string): string {
return `CHK_${tablePrefix}${table}_${column}`;
}

10
packages/@n8n/db/src/migrations/sqlite/1791542683581-AddTimeoutSecondsToScheduler.ts Normal file
View File

@ -0,0 +1,10 @@
import { AddTimeoutSecondsToScheduler1791542683581 as BaseMigration } from '../common/1791542683581-AddTimeoutSecondsToScheduler';
/**
* Only the rollback needs this: on SQLite, dropping a CHECK rebuilds the table.
* `scheduled_task` references `scheduled_job` with ON DELETE CASCADE, so the
* rebuild's DROP would wipe queued tasks. Disable foreign keys for the migration.
*/
export class AddTimeoutSecondsToScheduler1791542683581 extends BaseMigration {
withFKsDisabled = true as const;
}

208
packages/cli/test/migration/1791542683581-add-timeout-seconds-to-scheduler.test.ts Normal file
View File

@ -0,0 +1,208 @@
import {
createTestMigrationContext,
initDbUpToMigration,
runSingleMigration,
undoLastSingleMigration,
type TestMigrationContext,
} from '@n8n/backend-test-utils';
import { DbConnection } from '@n8n/db';
import { Container } from '@n8n/di';
import { DataSource, TableCheck } from '@n8n/typeorm';
const MIGRATION_NAME = 'AddTimeoutSecondsToScheduler1791542683581';
const isSqlite = (process.env.DB_TYPE ?? 'sqlite') === 'sqlite';
describe('AddTimeoutSecondsToScheduler Migration', () => {
let dataSource: DataSource;
beforeAll(async () => {
const dbConnection = Container.get(DbConnection);
await dbConnection.init();
dataSource = Container.get(DataSource);
});
beforeEach(async () => {
const context = createTestMigrationContext(dataSource);
await context.queryRunner.clearDatabase();
await context.queryRunner.release();
await initDbUpToMigration(MIGRATION_NAME);
});
afterAll(async () => {
const dbConnection = Container.get(DbConnection);
await dbConnection.close();
});
async function insertJob(context: TestMigrationContext, name: string, timeoutSeconds?: number) {
const column = timeoutSeconds === undefined ? '' : ', "timeoutSeconds"';
const value = timeoutSeconds === undefined ? '' : `, ${timeoutSeconds}`;
await context.runQuery(
`INSERT INTO ${context.escape.tableName('scheduled_job')}
("name", "kind", "intervalSeconds", "taskType", "ownerType", "ownerId", "createdAt", "updatedAt"${column})
VALUES ('${name}', 'interval', 60, 'test', 'workflow', 'wf-1', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP${value})`,
);
}
async function insertTaskFor(
context: TestMigrationContext,
jobName: string,
timeoutSeconds?: number,
) {
const [job] = (await context.queryRunner.query(
`SELECT "id" FROM ${context.escape.tableName('scheduled_job')} WHERE "name" = '${jobName}'`,
)) as Array<{ id: number }>;
const column = timeoutSeconds === undefined ? '' : ', "timeoutSeconds"';
const value = timeoutSeconds === undefined ? '' : `, ${timeoutSeconds}`;
await context.runQuery(
`INSERT INTO ${context.escape.tableName('scheduled_task')}
("jobId", "taskType", "scheduledFor", "runAt", "createdAt"${column})
VALUES (${job.id}, 'test', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP${value})`,
);
}
async function readTimeouts(context: TestMigrationContext, table: string): Promise<number[]> {
const rows = (await context.queryRunner.query(
`SELECT "timeoutSeconds" AS "timeout" FROM ${context.escape.tableName(table)}`,
)) as Array<{ timeout: number }>;
return rows.map((row) => Number(row.timeout));
}
async function countRows(context: TestMigrationContext, table: string): Promise<number> {
const [{ count }] = (await context.queryRunner.query(
`SELECT COUNT(*) AS "count" FROM ${context.escape.tableName(table)}`,
)) as Array<{ count: number | string }>;
return Number(count);
}
async function columnNames(context: TestMigrationContext, table: string): Promise<string[]> {
if (context.isSqlite) {
const rows = (await context.queryRunner.query(
`PRAGMA table_info(${context.escape.tableName(table)})`,
)) as Array<{ name: string }>;
return rows.map((row) => row.name);
}
const rows = (await context.queryRunner.query(
'SELECT column_name FROM information_schema.columns WHERE table_name = $1',
[`${context.tablePrefix}${table}`],
)) as Array<{ column_name: string }>;
return rows.map((row) => row.column_name);
}
describe('up', () => {
it('gives existing jobs and tasks the default timeout', async () => {
const seeded = createTestMigrationContext(dataSource);
await insertJob(seeded, 'job');
await insertTaskFor(seeded, 'job');
await seeded.queryRunner.release();
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
expect(await readTimeouts(context, 'scheduled_job')).toEqual([300]);
expect(await readTimeouts(context, 'scheduled_task')).toEqual([300]);
await context.queryRunner.release();
});
it.each([1, 2147483])('stores a timeout of %s', async (timeoutSeconds) => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
await insertJob(context, 'job', timeoutSeconds);
expect(await readTimeouts(context, 'scheduled_job')).toEqual([timeoutSeconds]);
await context.queryRunner.release();
});
it.each([0, -1, 2147484])('rejects a timeout of %s', async (timeoutSeconds) => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
await expect(insertJob(context, 'job', timeoutSeconds)).rejects.toThrow();
await context.queryRunner.release();
});
// Postgres coerces the value to `int` before the CHECK runs. SQLite stores it as given.
it.skipIf(!isSqlite)('rejects a fractional timeout on SQLite', async () => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
await expect(insertJob(context, 'job', 1.5)).rejects.toThrow();
await context.queryRunner.release();
});
it.each([0, -1, 2147484])('rejects a task timeout of %s', async (timeoutSeconds) => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
await insertJob(context, 'job');
await expect(insertTaskFor(context, 'job', timeoutSeconds)).rejects.toThrow();
await context.queryRunner.release();
});
it.skipIf(!isSqlite)('rejects a fractional task timeout on SQLite', async () => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
await insertJob(context, 'job');
await expect(insertTaskFor(context, 'job', 1.5)).rejects.toThrow();
await context.queryRunner.release();
});
});
describe('down', () => {
it('drops the columns without deleting a job or a queued task', async () => {
await runSingleMigration(MIGRATION_NAME);
const seeded = createTestMigrationContext(dataSource);
await insertJob(seeded, 'job', 600);
await insertTaskFor(seeded, 'job');
await seeded.queryRunner.release();
await undoLastSingleMigration();
const context = createTestMigrationContext(dataSource);
expect(await columnNames(context, 'scheduled_job')).not.toContain('timeoutSeconds');
expect(await columnNames(context, 'scheduled_task')).not.toContain('timeoutSeconds');
expect(await countRows(context, 'scheduled_job')).toBe(1);
expect(await countRows(context, 'scheduled_task')).toBe(1);
await context.queryRunner.release();
});
// On SQLite, a TypeORM rebuild turns the column CHECK into a table-level one,
// and SQLite refuses to drop a column that a table-level CHECK names.
it('drops the column after a later migration rebuilt the table', async () => {
await runSingleMigration(MIGRATION_NAME);
const rebuilt = createTestMigrationContext(dataSource);
const taskTable = `${rebuilt.tablePrefix}scheduled_task`;
await rebuilt.queryRunner.getTable(taskTable);
await rebuilt.queryRunner.createCheckConstraint(
taskTable,
new TableCheck({ name: 'CHK_rebuild_test', expression: '"attempts" >= 0' }),
);
await rebuilt.queryRunner.release();
await undoLastSingleMigration();
const context = createTestMigrationContext(dataSource);
expect(await columnNames(context, 'scheduled_task')).not.toContain('timeoutSeconds');
await context.queryRunner.release();
});
});
// NOT VALID skips the scan of the existing rows, and still checks every new one.
it.skipIf(isSqlite)('adds the CHECK on Postgres without validating existing rows', async () => {
await runSingleMigration(MIGRATION_NAME);
const context = createTestMigrationContext(dataSource);
const rows = (await context.queryRunner.query(
'SELECT convalidated FROM pg_constraint WHERE conname = $1',
[`CHK_${context.tablePrefix}scheduled_task_timeoutSeconds`],
)) as Array<{ convalidated: boolean }>;
expect(rows).toEqual([{ convalidated: false }]);
await context.queryRunner.release();
});
});