feat(core): Add a concurrency limit option to durable scheduled jobs (no-changelog) (#39166)

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

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

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

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

docs/generated/sqlite-schema/scheduled_job.md

@@ -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), 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'))))))
 ```
 
 </details>
@@ -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

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

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;

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)}`,
+		);
+	}
+}

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 });
+	}
+}

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);

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<ScheduledJob> {
 	async updateRunOptions(
 		manager: EntityManager,
 		ids: number[],
-		update: Pick<ScheduledJob, 'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds'>,
+		update: Pick<
+			ScheduledJob,
+			'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit'
+		>,
 	): Promise<void> {
 		if (ids.length === 0) return;
 		await manager.update(ScheduledJob, ids, update);

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',

packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts

@@ -23,6 +23,7 @@ const makeJob = (overrides: Partial<ScheduledJob> = {}): ScheduledJob => ({
 	nextRunAt: NOW,
 	lastFiredAt: null,
 	maxAttempts: 1,
+	concurrencyLimit: null,
 	misfirePolicy: ScheduledJobMisfirePolicy.CoalesceOwner,
 	misfireGraceSeconds: 60,
 	ownerKey: 'owner-a',

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',

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',

packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts

@@ -23,6 +23,7 @@ const makeJob = (overrides: Partial<ScheduledJob> = {}): ScheduledJob => ({
 	nextRunAt: NOW,
 	lastFiredAt: null,
 	maxAttempts: 1,
+	concurrencyLimit: null,
 	misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
 	misfireGraceSeconds: 60,
 	ownerKey: 'owner-1',

packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts

@@ -22,6 +22,7 @@ function makeJob(overrides: Partial<ScheduledJob> = {}): ScheduledJob {
 		nextRunAt: new Date('2026-01-01T00:00:00.000Z'),
 		lastFiredAt: null,
 		maxAttempts: 1,
+		concurrencyLimit: null,
 		misfirePolicy: 'coalesce',
 		misfireGraceSeconds: 60,
 		ownerKey: 'owner-1',

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. */

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<ScheduledJob> = {}): 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<ProvisionSummary> =>
+		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,
 				},
 			]);

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<ProvisionSummary> {
@@ -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;
+}

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,
 					}),
 			);
 

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),

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<number | null> {
+		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<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 indexExists(context: TestMigrationContext, name: string): Promise<boolean> {
+		const rows = context.isSqlite
+			? await context.runQuery<unknown[]>(
+					"SELECT name FROM sqlite_master WHERE type = 'index' AND name = :name",
+					{ name },
+				)
+			: await context.runQuery<unknown[]>(
+					'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<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.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();
+		});
+	});
+});