From 1f27b98366cda4186e3c2dc478bcced8648cfe00 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Mon, 12 Oct 2026 08:08:45 +0000 Subject: [PATCH] feat(core): Add a timeout column to the scheduler tables (no-changelog) (#40510) --- docs/generated/postgres-schema/README.md | 6 +- .../postgres-schema/public.scheduled_job.md | 5 + .../postgres-schema/public.scheduled_task.md | 5 + docs/generated/sqlite-schema/README.md | 6 +- docs/generated/sqlite-schema/scheduled_job.md | 6 +- .../generated/sqlite-schema/scheduled_task.md | 6 +- ...1542683581-AddTimeoutSecondsToScheduler.ts | 82 +++++++ ...1542683581-AddTimeoutSecondsToScheduler.ts | 10 + ...1-add-timeout-seconds-to-scheduler.test.ts | 208 ++++++++++++++++++ 9 files changed, 328 insertions(+), 6 deletions(-) create mode 100644 packages/@n8n/db/src/migrations/common/1791542683581-AddTimeoutSecondsToScheduler.ts create mode 100644 packages/@n8n/db/src/migrations/sqlite/1791542683581-AddTimeoutSecondsToScheduler.ts create mode 100644 packages/cli/test/migration/1791542683581-add-timeout-seconds-to-scheduler.test.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 18a776a05ab..cdb7a6648cb 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -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 diff --git a/docs/generated/postgres-schema/public.scheduled_job.md b/docs/generated/postgres-schema/public.scheduled_job.md index 38f157b98f3..36b5f799b3d 100644 --- a/docs/generated/postgres-schema/public.scheduled_job.md +++ b/docs/generated/postgres-schema/public.scheduled_job.md @@ -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 } ``` diff --git a/docs/generated/postgres-schema/public.scheduled_task.md b/docs/generated/postgres-schema/public.scheduled_task.md index 4ed375d9976..d6abf8fc32c 100644 --- a/docs/generated/postgres-schema/public.scheduled_task.md +++ b/docs/generated/postgres-schema/public.scheduled_task.md @@ -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 } diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index a61a4b9d929..c62c719c3f0 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -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 diff --git a/docs/generated/sqlite-schema/scheduled_job.md b/docs/generated/sqlite-schema/scheduled_job.md index 1f884e440ee..1bc9840ca7b 100644 --- a/docs/generated/sqlite-schema/scheduled_job.md +++ b/docs/generated/sqlite-schema/scheduled_job.md @@ -6,7 +6,7 @@ Table Definition ```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')))))) ``` @@ -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 } ``` diff --git a/docs/generated/sqlite-schema/scheduled_task.md b/docs/generated/sqlite-schema/scheduled_task.md index d2defcc9124..c9d8b43e1da 100644 --- a/docs/generated/sqlite-schema/scheduled_task.md +++ b/docs/generated/sqlite-schema/scheduled_task.md @@ -6,7 +6,7 @@ Table Definition ```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) ``` @@ -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 } diff --git a/packages/@n8n/db/src/migrations/common/1791542683581-AddTimeoutSecondsToScheduler.ts b/packages/@n8n/db/src/migrations/common/1791542683581-AddTimeoutSecondsToScheduler.ts new file mode 100644 index 00000000000..809246cfb7d --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1791542683581-AddTimeoutSecondsToScheduler.ts @@ -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}`; +} diff --git a/packages/@n8n/db/src/migrations/sqlite/1791542683581-AddTimeoutSecondsToScheduler.ts b/packages/@n8n/db/src/migrations/sqlite/1791542683581-AddTimeoutSecondsToScheduler.ts new file mode 100644 index 00000000000..9d622ae6b0a --- /dev/null +++ b/packages/@n8n/db/src/migrations/sqlite/1791542683581-AddTimeoutSecondsToScheduler.ts @@ -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; +} diff --git a/packages/cli/test/migration/1791542683581-add-timeout-seconds-to-scheduler.test.ts b/packages/cli/test/migration/1791542683581-add-timeout-seconds-to-scheduler.test.ts new file mode 100644 index 00000000000..bef28774608 --- /dev/null +++ b/packages/cli/test/migration/1791542683581-add-timeout-seconds-to-scheduler.test.ts @@ -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 { + 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 { + 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 { + 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(); + }); +});