diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 82464d12cb2..e3a273d45a4 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -111,7 +111,7 @@ 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) | 23 | | BASE TABLE | +| [public.scheduled_job](public.scheduled_job.md) | 24 | | BASE TABLE | | [public.scheduled_task](public.scheduled_task.md) | 18 | | BASE TABLE | | [public.scope](public.scope.md) | 3 | | BASE TABLE | | [public.secrets_provider_connection](public.secrets_provider_connection.md) | 7 | | BASE TABLE | @@ -1401,6 +1401,7 @@ erDiagram varchar_128_ scopeSlug FK } "public.scheduled_job" { + integer concurrencyLimit timestamp_3__with_time_zone createdAt varchar_255_ cronExpression boolean enabled diff --git a/docs/generated/postgres-schema/public.scheduled_job.md b/docs/generated/postgres-schema/public.scheduled_job.md index 9fad1110f2f..38f157b98f3 100644 --- a/docs/generated/postgres-schema/public.scheduled_job.md +++ b/docs/generated/postgres-schema/public.scheduled_job.md @@ -4,6 +4,7 @@ | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| concurrencyLimit | integer | | true | | | How many occurrences of this job may run at the same time. NULL means no limit. | | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | | cronExpression | varchar(255) | | true | | | Cron expression. For kind 'cron' it is the schedule; for 'recurring_cron' it lists the candidate run times that the every-N-periods filter then keeps every Nth of. | | enabled | boolean | true | false | | | Whether the scheduler considers this job for firing. | @@ -32,6 +33,7 @@ | Name | Type | Definition | | ---- | ---- | ---------- | +| CHK_scheduled_job_concurrencyLimit | CHECK | CHECK ((("concurrencyLimit" IS NULL) OR (("concurrencyLimit" >= 1) AND ("concurrencyLimit" <= 2147483647)))) | | CHK_scheduled_job_cron_expression | CHECK | CHECK ((((kind)::text <> 'cron'::text) OR ("cronExpression" IS NOT NULL))) | | CHK_scheduled_job_fire_at | CHECK | CHECK ((((kind)::text <> 'one_off'::text) OR ("fireAt" IS NOT NULL))) | | CHK_scheduled_job_interval_seconds | CHECK | CHECK ((((kind)::text <> 'interval'::text) OR ("intervalSeconds" IS NOT NULL))) | @@ -60,6 +62,7 @@ | Name | Definition | | ---- | ---------- | +| IDX_scheduled_job_concurrencyLimit | CREATE INDEX "IDX_scheduled_job_concurrencyLimit" ON public.scheduled_job USING btree ("concurrencyLimit") WHERE ("concurrencyLimit" IS NOT NULL) | | IDX_scheduled_job_name | CREATE UNIQUE INDEX "IDX_scheduled_job_name" ON public.scheduled_job USING btree (name) | | IDX_scheduled_job_nextRunAt | CREATE INDEX "IDX_scheduled_job_nextRunAt" ON public.scheduled_job USING btree ("nextRunAt") WHERE ((enabled = true) AND ("nextRunAt" IS NOT NULL)) | | IDX_scheduled_job_ownerType_ownerId_ownerMemberId | CREATE INDEX "IDX_scheduled_job_ownerType_ownerId_ownerMemberId" ON public.scheduled_job USING btree ("ownerType", "ownerId", "ownerMemberId") | @@ -73,6 +76,7 @@ erDiagram "public.scheduled_task" }o--|| "public.scheduled_job" : "FOREIGN KEY (#quot;jobId#quot;) REFERENCES scheduled_job(id) ON DELETE CASCADE" "public.scheduled_job" { + integer concurrencyLimit timestamp_3__with_time_zone createdAt varchar_255_ cronExpression boolean enabled diff --git a/docs/generated/postgres-schema/public.scheduled_task.md b/docs/generated/postgres-schema/public.scheduled_task.md index b3383e2ce21..4ed375d9976 100644 --- a/docs/generated/postgres-schema/public.scheduled_task.md +++ b/docs/generated/postgres-schema/public.scheduled_task.md @@ -82,6 +82,7 @@ erDiagram varchar_128_ taskType } "public.scheduled_job" { + integer concurrencyLimit timestamp_3__with_time_zone createdAt varchar_255_ cronExpression boolean enabled diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 0c95ff66aec..8539bf8e0d7 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -111,7 +111,7 @@ 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) | 23 | | table | +| [scheduled_job](scheduled_job.md) | 24 | | table | | [scheduled_task](scheduled_task.md) | 18 | | table | | [scope](scope.md) | 3 | | table | | [secrets_provider_connection](secrets_provider_connection.md) | 7 | | table | @@ -1387,6 +1387,7 @@ erDiagram VARCHAR_128_ scopeSlug PK } "scheduled_job" { + INT concurrencyLimit datetime_3_ createdAt varchar_255_ cronExpression boolean enabled diff --git a/docs/generated/sqlite-schema/scheduled_job.md b/docs/generated/sqlite-schema/scheduled_job.md index 7b03a6a69a9..1f884e440ee 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), 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")), 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')))))) ``` @@ -15,6 +15,7 @@ CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar( | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| concurrencyLimit | INT | | true | | | | | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | | cronExpression | varchar(255) | | true | | | | | enabled | boolean | true | false | | | | @@ -43,6 +44,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 ((((("misfireGraceSeconds" > 0))))) | | - | CHECK | CHECK (((((("recurrenceUnit" IN ('hours', 'days', 'weeks', 'months'))))))) | | - | CHECK | CHECK (((((("recurrenceSize" >= 2)))))) | @@ -58,6 +60,7 @@ CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar( | Name | Definition | | ---- | ---------- | +| IDX_scheduled_job_concurrencyLimit | CREATE INDEX "IDX_scheduled_job_concurrencyLimit" ON "scheduled_job" ("concurrencyLimit") WHERE "concurrencyLimit" IS NOT NULL | | IDX_scheduled_job_name | CREATE UNIQUE INDEX "IDX_scheduled_job_name" ON "scheduled_job" ("name") | | IDX_scheduled_job_nextRunAt | CREATE INDEX "IDX_scheduled_job_nextRunAt" ON "scheduled_job" ("nextRunAt") WHERE "enabled" = true AND "nextRunAt" IS NOT NULL | | IDX_scheduled_job_ownerType_ownerId_ownerMemberId | CREATE INDEX "IDX_scheduled_job_ownerType_ownerId_ownerMemberId" ON "scheduled_job" ("ownerType", "ownerId", "ownerMemberId") | @@ -70,6 +73,7 @@ erDiagram "scheduled_task" }o--|| "scheduled_job" : "FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "scheduled_job" { + INT concurrencyLimit datetime_3_ createdAt varchar_255_ cronExpression boolean enabled diff --git a/docs/generated/sqlite-schema/scheduled_task.md b/docs/generated/sqlite-schema/scheduled_task.md index 78de941adc0..d2defcc9124 100644 --- a/docs/generated/sqlite-schema/scheduled_task.md +++ b/docs/generated/sqlite-schema/scheduled_task.md @@ -81,6 +81,7 @@ erDiagram varchar_128_ taskType } "scheduled_job" { + INT concurrencyLimit datetime_3_ createdAt varchar_255_ cronExpression boolean enabled diff --git a/packages/@n8n/db/src/entities/scheduled-job.ts b/packages/@n8n/db/src/entities/scheduled-job.ts index 36c05c7d4a0..93e71968950 100644 --- a/packages/@n8n/db/src/entities/scheduled-job.ts +++ b/packages/@n8n/db/src/entities/scheduled-job.ts @@ -32,6 +32,9 @@ export { ScheduledJobMisfirePolicy, ScheduledJobOwnerType } from '@n8n/constants // leftmost prefix, the third column covers the per-member provisioning diff. @Index(['ownerType', 'ownerId', 'ownerMemberId']) @Index(['name'], { unique: true }) +// The claim reads the set of limited jobs on every pass. Partial, so it stays as +// small as that set. +@Index(['concurrencyLimit'], { where: '"concurrencyLimit" IS NOT NULL' }) export class ScheduledJob extends WithTimestamps { @PrimaryGeneratedColumn() id: number; @@ -155,6 +158,16 @@ export class ScheduledJob extends WithTimestamps { @Column({ type: 'int', default: 1 }) maxAttempts: number; + /** + * How many of this job's occurrences may run at the same time. `null` means no + * limit. + * + * At least 1: a lower ceiling would hold every occurrence back until its misfire + * deadline passed, so the job would never run. + */ + @Column({ type: 'int', nullable: true }) + concurrencyLimit: number | null; + /** What happens to an occurrence overdue by more than {@link misfireGraceSeconds}. */ @Column({ type: 'varchar', length: 16, default: ScheduledJobMisfirePolicy.Coalesce }) misfirePolicy: ScheduledJobMisfirePolicy; diff --git a/packages/@n8n/db/src/migrations/postgresdb/1790073725529-AddConcurrencyLimitToScheduledJob.ts b/packages/@n8n/db/src/migrations/postgresdb/1790073725529-AddConcurrencyLimitToScheduledJob.ts new file mode 100644 index 00000000000..80c2f0e136e --- /dev/null +++ b/packages/@n8n/db/src/migrations/postgresdb/1790073725529-AddConcurrencyLimitToScheduledJob.ts @@ -0,0 +1,39 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const table = 'scheduled_job'; +const column = 'concurrencyLimit'; + +const maxInt = 2147483647; + +/** + * Adds the per-job concurrency ceiling to `scheduled_job`, with a partial index + * over the jobs that carry one. NULL means no limit, so existing rows need no backfill. + */ +export class AddConcurrencyLimitToScheduledJob1790073725529 implements ReversibleMigration { + async up({ runQuery, escape, tablePrefix, schemaBuilder: { createIndex } }: MigrationContext) { + const tableName = escape.tableName(table); + const columnName = escape.columnName(column); + + // Limits below one would block all runs. + await runQuery( + `ALTER TABLE ${tableName} ADD COLUMN ${columnName} int ` + + `CONSTRAINT "CHK_${tablePrefix}${table}_${column}" CHECK (${columnName} IS NULL OR ` + + `(${columnName} >= 1 AND ${columnName} <= ${maxInt}))`, + ); + + await runQuery( + `COMMENT ON COLUMN ${tableName}.${columnName} IS ` + + "'How many occurrences of this job may run at the same time. NULL means no limit.'", + ); + + // Each task claim scans limited jobs. Exclude unlimited jobs to keep the index small. + await createIndex(table, [column], false, undefined, `${columnName} IS NOT NULL`); + } + + async down({ runQuery, escape, schemaBuilder }: MigrationContext) { + await schemaBuilder.dropIndex(table, [column]); + await runQuery( + `ALTER TABLE ${escape.tableName(table)} DROP COLUMN ${escape.columnName(column)}`, + ); + } +} diff --git a/packages/@n8n/db/src/migrations/sqlite/1790073725529-AddConcurrencyLimitToScheduledJob.ts b/packages/@n8n/db/src/migrations/sqlite/1790073725529-AddConcurrencyLimitToScheduledJob.ts new file mode 100644 index 00000000000..46bcc153842 --- /dev/null +++ b/packages/@n8n/db/src/migrations/sqlite/1790073725529-AddConcurrencyLimitToScheduledJob.ts @@ -0,0 +1,42 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const table = 'scheduled_job'; +const column = 'concurrencyLimit'; + +const maxInt = 2147483647; + +/** + * Adds the per-job concurrency ceiling to `scheduled_job`, with a partial index + * over the jobs that carry one. NULL means no limit, so existing rows need no backfill. + */ +export class AddConcurrencyLimitToScheduledJob1790073725529 implements ReversibleMigration { + // Rollback rebuilds scheduled_job. Disable foreign keys to preserve scheduled_task rows. + withFKsDisabled = true as const; + + async up({ runQuery, escape, tablePrefix, schemaBuilder: { createIndex } }: MigrationContext) { + const tableName = escape.tableName(table); + const columnName = escape.columnName(column); + + // Limits below one would block all runs. CAST rejects fractional limits, which an + // `int` column stores as-is here. The CHECK takes the name the DSL would give it, + // so down() can find it with dropCheckConstraint. + await runQuery( + `ALTER TABLE ${tableName} ADD COLUMN ${columnName} int ` + + `CONSTRAINT "CHK_${tablePrefix}${table}_${column}" CHECK (${columnName} IS NULL OR ` + + `(${columnName} >= 1 AND ${columnName} <= ${maxInt} AND CAST(${columnName} AS INTEGER) = ${columnName}))`, + ); + + // Each task claim scans limited jobs. Exclude unlimited jobs to keep the index small. + await createIndex(table, [column], false, undefined, `${columnName} IS NOT NULL`); + } + + async down({ queryRunner, schemaBuilder, tablePrefix }: MigrationContext) { + await schemaBuilder.dropIndex(table, [column]); + // Drop the CHECK first so the rebuild cannot reference the removed column. + await queryRunner.dropCheckConstraint( + `${tablePrefix}${table}`, + `CHK_${tablePrefix}${table}_${column}`, + ); + await schemaBuilder.dropColumns(table, [column], { recreatesOnSqlite: true }); + } +} diff --git a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts index d588f5c82ec..d62ae848599 100644 --- a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts @@ -30,6 +30,7 @@ const newJob = (name: string): NewScheduledJob => ({ maxAttempts: 5, misfirePolicy: 'coalesce', misfireGraceSeconds: 60, + concurrencyLimit: null, }); /** A row as the post-insert read-back returns it: id, name and the owner it belongs to. */ @@ -418,6 +419,7 @@ describe('ScheduledJobRepository', () => { maxAttempts: 3, misfirePolicy: 'skip', misfireGraceSeconds: 120, + concurrencyLimit: 2, }; await repository.updateDefinition(entityManager, 10, update); diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts index 7fb9de6c758..795bab6c618 100644 --- a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -37,6 +37,7 @@ export type NewScheduledJob = Pick< | 'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds' + | 'concurrencyLimit' >; /** A changed schedule definition, plus the fresh clock it restarts from. */ @@ -53,6 +54,7 @@ export type ScheduledJobDefinitionUpdate = Pick< | 'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds' + | 'concurrencyLimit' >; @Service() @@ -268,7 +270,10 @@ export class ScheduledJobRepository extends Repository { async updateRunOptions( manager: EntityManager, ids: number[], - update: Pick, + update: Pick< + ScheduledJob, + 'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit' + >, ): Promise { if (ids.length === 0) return; await manager.update(ScheduledJob, ids, update); diff --git a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts b/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts index 9fe064c063b..5afe5006e3e 100644 --- a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts +++ b/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts @@ -355,6 +355,7 @@ describe('createScheduler materialize', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -401,6 +402,7 @@ describe('createScheduler materialize', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -1265,6 +1267,7 @@ describe('createScheduler tracing', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -1318,6 +1321,7 @@ describe('createScheduler tracing', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -1614,6 +1618,7 @@ describe('createScheduler metrics', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts index 98934fa3d49..d912cc12aeb 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts @@ -23,6 +23,7 @@ const makeJob = (overrides: Partial = {}): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.CoalesceOwner, misfireGraceSeconds: 60, ownerKey: 'owner-a', diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts index fc89b371ea5..1674d53a06c 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts @@ -22,6 +22,7 @@ const makeJob = (id: number): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts index 4655caf8eec..4fa23c9d687 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts @@ -21,6 +21,7 @@ const makeIntervalJob = (intervalSeconds: number): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -40,6 +41,7 @@ const makeCronJob = (cronExpression: string, timezone: string): ScheduledJob => nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', @@ -61,6 +63,7 @@ const makeOneOffJob = (fireAt: Date): ScheduledJob => ({ nextRunAt: fireAt, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts index 051d648a3e5..e6e2a477004 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts @@ -23,6 +23,7 @@ const makeJob = (overrides: Partial = {}): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, ownerKey: 'owner-1', diff --git a/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts b/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts index 7e9abb5e74e..7eb2258bbab 100644 --- a/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts +++ b/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts @@ -22,6 +22,7 @@ function makeJob(overrides: Partial = {}): ScheduledJob { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 1, + concurrencyLimit: null, misfirePolicy: 'coalesce', misfireGraceSeconds: 60, ownerKey: 'owner-1', diff --git a/packages/@n8n/scheduler/src/core/types/schedule.ts b/packages/@n8n/scheduler/src/core/types/schedule.ts index 20b2b2cac05..a3cc1b3bc15 100644 --- a/packages/@n8n/scheduler/src/core/types/schedule.ts +++ b/packages/@n8n/scheduler/src/core/types/schedule.ts @@ -103,6 +103,11 @@ export interface ScheduledJob { nextRunAt: Date | null; // the next instant the materializer materializes from. lastFiredAt: Date | null; maxAttempts: number; + /** + * How many of this job's occurrences may run at the same time. `null` means no + * limit. At least 1 when set. + */ + concurrencyLimit: number | null; /** What to do with occurrences that came due while nothing ran them. */ misfirePolicy: ScheduledJobMisfirePolicy; /** How late an occurrence may be before {@link misfirePolicy} applies to it. */ diff --git a/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts b/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts index e0d540bd1b3..c378be661b6 100644 --- a/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts +++ b/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts @@ -10,6 +10,7 @@ import type { import type { DesiredJob, ProvisionSummary, ScheduleDefinition } from '@n8n/scheduler'; import type { EntityManager } from '@n8n/typeorm'; import type { Tracing } from 'n8n-core'; +import { UserError } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; import { DurableJobProvisioner } from '../durable-job-provisioner'; @@ -54,6 +55,7 @@ const jobRow = ({ payload = {}, ...over }: Partial = {}): Schedule maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, ...over, }), { payload }, @@ -140,6 +142,19 @@ describe('DurableJobProvisioner', () => { maxAttempts, }); + const provisionWithConcurrencyLimit = async ( + concurrencyLimit: number | null | undefined, + desired: DesiredJob[] = [desiredJob('wf:node:0')], + ): Promise => + await provisioner.provision({ + owner: OWNER, + taskType: 'schedule-trigger', + payload: {}, + desired, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + concurrencyLimit, + }); + beforeEach(() => { vi.resetAllMocks(); // Run the callback with our manager, standing in for a real transaction. @@ -190,6 +205,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, }, ]); expect(summary.inserted).toEqual([{ id: 100, name: 'wf:node:0' }]); @@ -257,6 +273,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 90, + concurrencyLimit: null, }); // The schedule is untouched, so the job keeps its queued tasks. expect(jobs.updateDefinition).not.toHaveBeenCalled(); @@ -292,6 +309,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, }); }); @@ -351,6 +369,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, }); expect(tasks.deletePendingByJobIds).toHaveBeenCalledWith(manager, [10]); expect(summary.redefined).toEqual([{ id: 10, name: 'wf:node:0' }]); @@ -645,6 +664,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, + concurrencyLimit: null, }); expect(tasks.updateMissedAfterForJobs).toHaveBeenCalledWith(manager, [10], 60); }); @@ -660,6 +680,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 300, + concurrencyLimit: null, }); expect(tasks.updateMissedAfterForJobs).toHaveBeenCalledWith(manager, [10], 300); }); @@ -709,6 +730,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 1, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, }); expect(jobs.updateDefinition).not.toHaveBeenCalled(); // The grace is unchanged, so queued tasks keep their deadline. @@ -724,6 +746,95 @@ describe('DurableJobProvisioner', () => { }); }); + describe('concurrency limit resolution', () => { + it('stamps a request-supplied limit onto the inserted row', async () => { + await provisionWithConcurrencyLimit(2); + + expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ + expect.objectContaining({ concurrencyLimit: 2 }), + ]); + }); + + it('leaves the inserted row unlimited when the request omits a limit', async () => { + await provisionWithConcurrencyLimit(undefined); + + expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ + expect.objectContaining({ concurrencyLimit: null }), + ]); + }); + + it("writes a request-supplied limit onto a redefined job's row", async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow()]); + + await provisionWithConcurrencyLimit(2, [ + desiredJob('wf:node:0', { + kind: 'cron', + cronExpression: '0 0 18 * * *', + timezone: 'UTC', + }), + ]); + + expect(jobs.updateDefinition).toHaveBeenCalledWith( + manager, + 10, + expect.objectContaining({ concurrencyLimit: 2 }), + ); + }); + + it('reconciles the limit of a job whose schedule is unchanged', async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow({ concurrencyLimit: 1 })]); + + await provisionWithConcurrencyLimit(2); + + expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { + maxAttempts: 5, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + misfireGraceSeconds: 90, + concurrencyLimit: 2, + }); + expect(jobs.updateDefinition).not.toHaveBeenCalled(); + }); + + it('lifts the limit of a job the request no longer limits', async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow({ concurrencyLimit: 2 })]); + + await provisionWithConcurrencyLimit(undefined); + + expect(jobs.updateRunOptions).toHaveBeenCalledWith( + manager, + [10], + expect.objectContaining({ concurrencyLimit: null }), + ); + }); + + it('leaves a job already stored at the requested limit out of the reconciliation', async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow({ concurrencyLimit: 2 })]); + + await provisionWithConcurrencyLimit(2); + + expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [], expect.anything()); + }); + + it('stores the largest limit the column holds', async () => { + await provisionWithConcurrencyLimit(2_147_483_647); + + expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ + expect.objectContaining({ concurrencyLimit: 2_147_483_647 }), + ]); + }); + + // 2_147_483_648 overflows the column on Postgres while SQLite would take it. + it.each([0, -1, 1.5, Number.NaN, 2_147_483_648])( + 'rejects a limit of %s', + async (concurrencyLimit) => { + await expect(provisionWithConcurrencyLimit(concurrencyLimit)).rejects.toThrow(UserError); + + expect(jobs.insertMany).not.toHaveBeenCalled(); + expect(dataSource.transaction).not.toHaveBeenCalled(); + }, + ); + }); + describe('seeding a freshly provisioned job', () => { const SEED_NOW = new Date('2026-01-05T00:00:00.000Z'); const at = (seconds: number) => new Date(SEED_NOW.getTime() + seconds * 1000); @@ -927,6 +1038,7 @@ describe('DurableJobProvisioner', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, + concurrencyLimit: null, ...columns, }, ]); diff --git a/packages/cli/src/scheduling/durable-job-provisioner.ts b/packages/cli/src/scheduling/durable-job-provisioner.ts index e6b2f6e57eb..3dc2d8eab46 100644 --- a/packages/cli/src/scheduling/durable-job-provisioner.ts +++ b/packages/cli/src/scheduling/durable-job-provisioner.ts @@ -1,6 +1,6 @@ import { Logger } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; -import { type ScheduledJobMisfirePolicy, Time } from '@n8n/constants'; +import { MAX_INTEGER_32BITS_SIGNED, type ScheduledJobMisfirePolicy, Time } from '@n8n/constants'; import type { EntityManager, NewScheduledJob, @@ -26,6 +26,7 @@ import type { RunInTransaction, } from '@n8n/scheduler'; import { Tracing } from 'n8n-core'; +import { UserError } from 'n8n-workflow'; import { isDeepStrictEqual } from 'node:util'; import { AgentScheduledJobOwner } from './agent-scheduled-job-owner'; @@ -41,6 +42,13 @@ import { WorkflowScheduledJobOwner } from './workflow-scheduled-job-owner'; */ const MAX_MISFIRE_GRACE_SECONDS = 30 * Time.days.toSeconds; +/** + * Ceiling for a concurrency limit: what the column's `int` holds. Postgres rejects + * anything above it outright, so it is checked here for both dialects to behave the + * same. + */ +const MAX_CONCURRENCY_LIMIT = MAX_INTEGER_32BITS_SIGNED; + /** One provisioning call: whose jobs to reconcile, and what to stamp on new rows. */ export interface ProvisionRequest { /** Who these jobs belong to. */ @@ -59,6 +67,11 @@ export interface ProvisionRequest { misfireGraceSeconds?: number; /** Retry ceiling stamped on each occurrence; omit to inherit the instance setting. */ maxAttempts?: number; + /** + * How many of the job's occurrences may run at the same time. Omit or pass + * `null` for no limit. + */ + concurrencyLimit?: number | null; } /** What provisioning stamps on the rows it writes, plus the owner it diffs against. */ @@ -161,6 +174,7 @@ export class DurableJobProvisioner { * liveness resolver (see {@link createJobProvisioner}). * @throws {InvalidOwnerIdError} when the owner id is empty or too long to store. * @throws {InvalidOwnerMemberIdError} when the owner member id is empty or too long to store. + * @throws {UserError} when the concurrency limit is not a whole number of at least 1. * @returns what the call inserted, redefined, left unchanged and removed. */ async provision({ desired, ...scope }: ProvisionRequest): Promise { @@ -237,12 +251,14 @@ export class DurableJobProvisioner { misfirePolicy, misfireGraceSeconds: requestedMisfireGraceSeconds, maxAttempts: requestedMaxAttempts, + concurrencyLimit: requestedConcurrencyLimit, }: ProvisionScope): RunInProvisionTransaction { const misfireGraceSeconds = this.resolveMisfireGraceSeconds( requestedMisfireGraceSeconds, owner, ); const maxAttempts = requestedMaxAttempts ?? this.globalConfig.scheduler.maxAttempts; + const concurrencyLimit = resolveConcurrencyLimit(requestedConcurrencyLimit); return async (work) => await this.dataSource.transaction(async (manager) => { // Provisioning is evidence the owner is back, so lift any quarantine now @@ -272,7 +288,8 @@ export class DurableJobProvisioner { if ( graceChanged || row.misfirePolicy !== misfirePolicy || - row.maxAttempts !== maxAttempts + row.maxAttempts !== maxAttempts || + row.concurrencyLimit !== concurrencyLimit ) { outdatedRunOptionJobIds.push(row.id); } @@ -301,6 +318,7 @@ export class DurableJobProvisioner { maxAttempts, misfirePolicy, misfireGraceSeconds, + concurrencyLimit, }), ); const ids = await this.jobs.insertMany(manager, rows); @@ -314,6 +332,7 @@ export class DurableJobProvisioner { maxAttempts, misfirePolicy, misfireGraceSeconds, + concurrencyLimit, }); seededJobIds.add(jobId); }, @@ -327,6 +346,7 @@ export class DurableJobProvisioner { maxAttempts, misfirePolicy, misfireGraceSeconds, + concurrencyLimit, }); // Only `insert` writes the payload, so an existing row picks up a change to it here. await this.jobs.updatePayload(manager, outdatedPayloadJobIds, payload); @@ -472,3 +492,23 @@ export class DurableJobProvisioner { } } } + +/** + * Normalize a requested concurrency ceiling to what the column stores: a whole + * number from 1 to {@link MAX_CONCURRENCY_LIMIT}, or `null` for no limit. + * + * @throws {UserError} when the limit is set but falls outside that range. A ceiling + * below one would hold every occurrence back until its misfire deadline passed, so + * the job would never run. + */ +function resolveConcurrencyLimit(requested: number | null | undefined): number | null { + if (requested === undefined || requested === null) { + return null; + } + if (!Number.isInteger(requested) || requested < 1 || requested > MAX_CONCURRENCY_LIMIT) { + throw new UserError('Scheduled job concurrency limit is outside the range the column holds', { + extra: { concurrencyLimit: requested, maxConcurrencyLimit: MAX_CONCURRENCY_LIMIT }, + }); + } + return requested; +} diff --git a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts index 57ac1dbefb8..78c1e2b3d15 100644 --- a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts +++ b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts @@ -90,6 +90,7 @@ describe('scheduled repositories', () => { name, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, + concurrencyLimit: null, ...selfOwned(name), taskType: 'scheduleTrigger', payload: {}, @@ -459,6 +460,17 @@ describe('scheduled repositories', () => { expect(ids).toEqual([idByName.get('wf:node:0'), idByName.get('wf:node:1')]); }); + it('stores a concurrency limit, and leaves an omitted one unlimited', async () => { + const jobs = [newJobRow('wf:node:0'), newJobRow('wf:node:1', { concurrencyLimit: 2 })]; + + await dataSource.transaction(async (trx) => await jobRepository.insertMany(trx, jobs)); + + const unlimited = await jobRepository.findOneByOrFail({ name: 'wf:node:0' }); + const limited = await jobRepository.findOneByOrFail({ name: 'wf:node:1' }); + expect(unlimited.concurrencyLimit).toBeNull(); + expect(limited.concurrencyLimit).toBe(2); + }); + it('returns the existing id for a name already taken, without duplicating the row', async () => { // A prior writer already holds `wf:node:0`; a second provisioning run inserts it // again alongside a fresh name. orIgnore skips the taken row, and the read-back by @@ -574,6 +586,7 @@ describe('scheduled repositories', () => { maxAttempts: 5, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 120, + concurrencyLimit: null, }), ); @@ -590,6 +603,23 @@ describe('scheduled repositories', () => { expect(other.misfireGraceSeconds).toBe(60); }); + it('rewrites the concurrency limit of the given jobs', async () => { + const job = await createJob({ concurrencyLimit: null }); + + await dataSource.transaction( + async (trx) => + await jobRepository.updateRunOptions(trx, [job.id], { + maxAttempts: job.maxAttempts, + misfirePolicy: job.misfirePolicy, + misfireGraceSeconds: job.misfireGraceSeconds, + concurrencyLimit: 3, + }), + ); + + const after = await jobRepository.findOneByOrFail({ id: job.id }); + expect(after.concurrencyLimit).toBe(3); + }); + it('leaves the queued occurrences of an updated job in place', async () => { const job = await createJob(); await createTask(job.id); @@ -600,6 +630,7 @@ describe('scheduled repositories', () => { maxAttempts: job.maxAttempts, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 60, + concurrencyLimit: null, }), ); @@ -615,6 +646,7 @@ describe('scheduled repositories', () => { maxAttempts: job.maxAttempts, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 120, + concurrencyLimit: null, }), ); diff --git a/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts b/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts index 9552f6c2ebf..89e8c740380 100644 --- a/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts +++ b/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts @@ -466,6 +466,7 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => { maxAttempts: 1, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, + concurrencyLimit: null, })); const ids = await dataSource.transaction( async (trx) => await jobRepository.insertMany(trx, newJobs), diff --git a/packages/cli/test/migration/1790073725529-add-concurrency-limit-to-scheduled-job.test.ts b/packages/cli/test/migration/1790073725529-add-concurrency-limit-to-scheduled-job.test.ts new file mode 100644 index 00000000000..21cf4f1dd79 --- /dev/null +++ b/packages/cli/test/migration/1790073725529-add-concurrency-limit-to-scheduled-job.test.ts @@ -0,0 +1,162 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + undoLastSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; + +const MIGRATION_NAME = 'AddConcurrencyLimitToScheduledJob1790073725529'; +const isSqlite = (process.env.DB_TYPE ?? 'sqlite') === 'sqlite'; + +describe('AddConcurrencyLimitToScheduledJob 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, + concurrencyLimit: number | null, + ) { + await context.runQuery( + `INSERT INTO ${context.escape.tableName('scheduled_job')} + ("name", "kind", "intervalSeconds", "taskType", "ownerType", "ownerId", "concurrencyLimit", "createdAt", "updatedAt") + VALUES ('${name}', 'interval', 60, 'test', 'workflow', 'wf-1', ${concurrencyLimit ?? 'NULL'}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`, + ); + } + + async function insertTaskFor(context: TestMigrationContext, jobName: string) { + const [job] = (await context.queryRunner.query( + `SELECT "id" FROM ${context.escape.tableName('scheduled_job')} WHERE "name" = '${jobName}'`, + )) as Array<{ id: number }>; + await context.runQuery( + `INSERT INTO ${context.escape.tableName('scheduled_task')} + ("jobId", "taskType", "scheduledFor", "runAt", "createdAt") + VALUES (${job.id}, 'test', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`, + ); + } + + async function readLimit(context: TestMigrationContext, name: string): Promise { + const [row] = (await context.queryRunner.query( + `SELECT "concurrencyLimit" AS "limit" FROM ${context.escape.tableName('scheduled_job')} WHERE "name" = '${name}'`, + )) as Array<{ limit: number | null }>; + return row.limit; + } + + 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 indexExists(context: TestMigrationContext, name: string): Promise { + const rows = context.isSqlite + ? await context.runQuery( + "SELECT name FROM sqlite_master WHERE type = 'index' AND name = :name", + { name }, + ) + : await context.runQuery( + 'SELECT indexname FROM pg_indexes WHERE indexname = :name', + { name }, + ); + return rows.length === 1; + } + + const limitIndexName = (context: TestMigrationContext) => + `IDX_${context.tablePrefix}scheduled_job_concurrencyLimit`; + + 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.each([null, 1, 2147483647])('stores a limit of %s', async (concurrencyLimit) => { + await runSingleMigration(MIGRATION_NAME); + const context = createTestMigrationContext(dataSource); + + await insertJob(context, 'job', concurrencyLimit); + + expect(await readLimit(context, 'job')).toBe(concurrencyLimit); + await context.queryRunner.release(); + }); + + it.each([0, -1, 2147483648])('rejects a limit of %s', async (concurrencyLimit) => { + await runSingleMigration(MIGRATION_NAME); + const context = createTestMigrationContext(dataSource); + + await expect(insertJob(context, 'job', concurrencyLimit)).rejects.toThrow(); + + await context.queryRunner.release(); + }); + + it('indexes the jobs that carry a limit', async () => { + await runSingleMigration(MIGRATION_NAME); + const context = createTestMigrationContext(dataSource); + + expect(await indexExists(context, limitIndexName(context))).toBe(true); + + await context.queryRunner.release(); + }); + + // Postgres has no fractional case to reject: the column type coerces the + // value before the CHECK runs. SQLite stores it as given. + it.skipIf(!isSqlite)('rejects a fractional limit on SQLite', async () => { + await runSingleMigration(MIGRATION_NAME); + const context = createTestMigrationContext(dataSource); + + await expect(insertJob(context, 'job', 1.5)).rejects.toThrow(); + + await context.queryRunner.release(); + }); + }); + + describe('down', () => { + it('drops the column without deleting a job or a queued task', async () => { + await runSingleMigration(MIGRATION_NAME); + const seeded = createTestMigrationContext(dataSource); + await insertJob(seeded, 'job', 2); + await insertTaskFor(seeded, 'job'); + await seeded.queryRunner.release(); + + await undoLastSingleMigration(); + + const context = createTestMigrationContext(dataSource); + expect(await indexExists(context, limitIndexName(context))).toBe(false); + expect(await columnNames(context, 'scheduled_job')).not.toContain('concurrencyLimit'); + expect(await countRows(context, 'scheduled_job')).toBe(1); + expect(await countRows(context, 'scheduled_task')).toBe(1); + await context.queryRunner.release(); + }); + }); +});