feat(core): Cap concurrent occurrences of a durable scheduled job (no-changelog) (#39197)

packages/@n8n/db/src/repositories/scheduled-task.repository.ts

@@ -123,6 +123,7 @@ export type ScheduledTaskMetricSnapshot = {
 export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	private readonly isPostgres: boolean;
 	private readonly tableName: string;
+	private readonly jobTableName: string;
 	// Quoted here so the reaper update below doesn't have to quote these
 	// camelCase columns itself for Postgres (SQLite accepts the same
 	// double-quoted identifier).
@@ -135,6 +136,7 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 		super(ScheduledTask, dataSource.manager);
 		this.isPostgres = config.type === 'postgresdb';
 		this.tableName = this.manager.connection.driver.escape(`${config.tablePrefix}scheduled_task`);
+		this.jobTableName = this.manager.connection.driver.escape(`${config.tablePrefix}scheduled_job`);
 		this.leaseExpiresAtColumn = this.manager.connection.driver.escape('leaseExpiresAt');
 		this.dispatchedAtColumn = this.manager.connection.driver.escape('dispatchedAt');
 		this.runAtColumn = this.manager.connection.driver.escape('runAt');
@@ -284,23 +286,67 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	 *
 	 * Rows past their deadline are left alone (see {@link claimableSql}).
 	 *
-	 * Concurrent claimers never take the same row: Postgres skips locked rows
-	 * (`FOR UPDATE SKIP LOCKED`); SQLite serialises via the sqlite-pooled driver's
-	 * `BEGIN IMMEDIATE`, which reserves the write lock upfront.
+	 * A job's `concurrencyLimit` caps how many of its tasks run at once. Extra tasks
+	 * stay `pending` for a later claim, and other jobs still fill the batch.
+	 *
+	 * Concurrent claimers never take the same row.
 	 */
 	async claimDueTasks(opts: ClaimDueTasksOptions): Promise<ScheduledTask[]> {
 		if (opts.taskTypes.length === 0) return [];
 		return this.isPostgres ? await this.claimWithPostgres(opts) : await this.claimWithSqlite(opts);
 	}
 
+	private claimCandidateSql(alias: string, taskTypesSql: string, dueBeforeSql: string): string {
+		return `${alias}"status" = '${ScheduledTaskStatus.Pending}'
+			AND ${alias}"taskType" ${taskTypesSql}
+			AND ${alias}"runAt" <= ${dueBeforeSql}
+			AND ${this.claimableSql(alias)}`;
+	}
+
+	private limitedJobIdsSql(): string {
+		return `SELECT "id" FROM ${this.jobTableName} WHERE "concurrencyLimit" IS NOT NULL`;
+	}
+
+	private withinConcurrencyLimitSql(alias: string): string {
+		return `(${alias}"jobId" NOT IN (${this.limitedJobIdsSql()})
+			OR ${alias}"id" IN (${this.allowedByConcurrencyLimitSql()}))`;
+	}
+
+	private allowedByConcurrencyLimitSql(): string {
+		return `
+			SELECT ranked."id"
+			FROM (
+				SELECT
+					c."id",
+					c."status",
+					ROW_NUMBER() OVER (
+						PARTITION BY c."jobId"
+						ORDER BY
+							CASE WHEN c."status" = '${ScheduledTaskStatus.Pending}' THEN 0 ELSE 1 END,
+							c."runAt",
+							c."id"
+					) AS "slot",
+					cj."concurrencyLimit" - SUM(
+						CASE WHEN c."status" = '${ScheduledTaskStatus.Running}' THEN 1 ELSE 0 END
+					) OVER (PARTITION BY c."jobId") AS "freeSlots"
+				FROM ${this.tableName} c
+				JOIN ${this.jobTableName} cj ON cj."id" = c."jobId"
+				WHERE cj."concurrencyLimit" IS NOT NULL
+					AND (
+						c."status" = '${ScheduledTaskStatus.Running}'
+						OR (c."status" = '${ScheduledTaskStatus.Pending}' AND ${this.claimableSql('c.')})
+					)
+			) ranked
+			WHERE ranked."status" = '${ScheduledTaskStatus.Pending}'
+				AND ranked."slot" <= ranked."freeSlots"
+		`;
+	}
+
 	/**
 	 * Whether a pending row is still worth claiming, as a SQL fragment for the given
 	 * table alias. A row past its `missedAfter` is left alone, unless it has no
 	 * deadline at all, or has already been attempted: a retry's backoff pushes it past
 	 * its deadline by design, and the attempt count decides its fate from then on.
-	 *
-	 * Shared by both dialects' claims and by the metric snapshot, so what the snapshot
-	 * calls due cannot drift from what the claim would take.
 	 */
 	private claimableSql(alias = ''): string {
 		const deadline = `${alias}"missedAfter"`;
@@ -308,22 +354,43 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	}
 
 	private async claimWithPostgres(opts: ClaimDueTasksOptions): Promise<ScheduledTask[]> {
+		const taskTypesSql = '= ANY($3)';
+		const dueBeforeSql = "now() + ($4 || ' milliseconds')::interval";
+
+		// Two claimers must not both use a job's last free place. The `xmin` guard skips
+		// a job that another claim changed, and `claimed_jobs` changes each job it uses.
 		// TypeORM's Postgres driver returns `[rows, affectedCount]` from a raw UPDATE
 		// ... RETURNING, so destructure the rows out of the tuple.
 		const [rows]: [ScheduledTask[], number] = await this.query(
-			`UPDATE ${this.tableName}
+			`WITH candidates AS MATERIALIZED (
+			   SELECT t."id", t."jobId", j."concurrencyLimit"
+			     FROM ${this.tableName} t
+			     JOIN ${this.jobTableName} j ON j."id" = t."jobId"
+			    WHERE ${this.claimCandidateSql('t.', taskTypesSql, dueBeforeSql)}
+			      AND ${this.withinConcurrencyLimitSql('t.')}
+			      AND (j."concurrencyLimit" IS NULL OR EXISTS (
+			        SELECT 1 FROM ${this.jobTableName} guard
+			         WHERE guard."id" = j."id" AND guard.xmin = j.xmin
+			         FOR NO KEY UPDATE OF guard SKIP LOCKED
+			      ))
+			    ORDER BY t."runAt"
+			    LIMIT $5
+			    FOR UPDATE OF t SKIP LOCKED
+			 ), claimed_jobs AS (
+			   UPDATE ${this.jobTableName} j
+			      SET "concurrencyLimit" = j."concurrencyLimit"
+			    WHERE j."id" IN (
+			      SELECT "jobId" FROM candidates WHERE "concurrencyLimit" IS NOT NULL
+			    )
+			   RETURNING j."id"
+			 )
+			 UPDATE ${this.tableName}
 			   SET "status" = '${ScheduledTaskStatus.Running}', "claimedBy" = $1,
 			       "leaseExpiresAt" = now() + ($2 || ' milliseconds')::interval,
 			       "leaseEpoch" = "leaseEpoch" + 1
 			 WHERE "id" IN (
-			   SELECT t."id" FROM ${this.tableName} t
-			    WHERE t."status" = '${ScheduledTaskStatus.Pending}'
-			      AND t."taskType" = ANY($3)
-			      AND t."runAt" <= now() + ($4 || ' milliseconds')::interval
-			      AND ${this.claimableSql('t.')}
-			    ORDER BY t."runAt"
-			    LIMIT $5
-			    FOR UPDATE SKIP LOCKED)
+			   SELECT "id" FROM candidates
+			    WHERE "concurrencyLimit" IS NULL OR "jobId" IN (SELECT "id" FROM claimed_jobs))
 			 RETURNING *`,
 			[opts.host, String(opts.leaseMs), opts.taskTypes, String(opts.lookaheadMs), opts.batchSize],
 		);
@@ -336,18 +403,18 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 		// so concurrent claimers serialise here rather than both reading the same
 		// pending rows. Two statements (select then update) because SQLite can't
 		// return the updated rows from an UPDATE.
+		const taskTypesSql = 'IN (:...taskTypes)';
+		// Reach `lookaheadMs` past now so a task due before the next poll is claimed
+		// early and fired precisely by the timer. STRFTIME takes the offset as whole
+		// seconds.
+		const dueBeforeSql = "STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', :lookahead)";
+		const params = { taskTypes: opts.taskTypes, lookahead: `+${opts.lookaheadMs / 1000} seconds` };
+
 		return await this.manager.transaction(async (tx) => {
 			const candidates = await tx
 				.createQueryBuilder(ScheduledTask, 't')
-				.where('t.status = :pending', { pending: ScheduledTaskStatus.Pending })
-				.andWhere('t.taskType IN (:...taskTypes)', { taskTypes: opts.taskTypes })
-				// Reach `lookaheadMs` past now so a task due before the next poll is claimed
-				// early and fired precisely by the timer. STRFTIME takes the offset as whole
-				// seconds.
-				.andWhere("t.runAt <= STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', :lookahead)", {
-					lookahead: `+${opts.lookaheadMs / 1000} seconds`,
-				})
-				.andWhere(this.claimableSql('t.'))
+				.where(this.claimCandidateSql('t.', taskTypesSql, dueBeforeSql), params)
+				.andWhere(this.withinConcurrencyLimitSql('t.'))
 				.orderBy('t.runAt', 'ASC')
 				.limit(opts.batchSize)
 				.getMany();
@@ -511,9 +578,10 @@ export class ScheduledTaskRepository extends Repository<ScheduledTask> {
 	 * (`oldestPendingAgeMs`). Everything reads the DB clock, so `due`-ness and the
 	 * oldest-age reference are consistent regardless of any instance clock skew.
 	 *
-	 * `due` and the oldest age count only what the claim would actually take, so a
+	 * `due` and the oldest age skip rows past their deadline, like the claim, so a
 	 * backlog the reaper has yet to retire doesn't read as work the scheduler is behind
-	 * on. `pending` stays the whole backlog, retirable rows included.
+	 * on. Rows a `concurrencyLimit` holds back still count as due. `pending` stays the
+	 * whole backlog, retirable rows included.
 	 *
 	 * The caller runs this behind a short scrape cache.
 	 */

packages/@n8n/scheduler/src/core/executor/store.ts

@@ -41,6 +41,9 @@ export interface ExecutorTaskStore {
 	/**
 	 * Atomically claim up to `batchSize` due tasks of the given types: mark them
 	 * `running`, owned by `host`, with a lease of `leaseMs` and a bumped epoch.
+	 *
+	 * A job's `concurrencyLimit` caps how many of its tasks run at once. Extra tasks
+	 * stay `pending`.
 	 */
 	claimDueTasks(batch: ClaimDueTasksBatch): Promise<ClaimedTask[]>;
 

packages/cli/test/integration/scheduling/executor-repository.test.ts

@@ -1,6 +1,7 @@
 import { testDb } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
 import type { ScheduledJob as ScheduledJobEntity, ScheduledTask } from '@n8n/db';
-import { ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
+import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
 
 import { selfOwned } from './shared/job-factory';
@@ -11,6 +12,7 @@ import { selfOwned } from './shared/job-factory';
  * Due-ness uses the DB clock, so tasks are made due with a `runAt` in the past.
  */
 describe('ScheduledTaskRepository executor methods', () => {
+	const isPostgres = process.env.DB_TYPE === 'postgresdb';
 	const TASK_TYPE = 'scheduleTrigger';
 	const HOST_A = 'main-a';
 	const HOST_B = 'main-b';
@@ -18,6 +20,7 @@ describe('ScheduledTaskRepository executor methods', () => {
 	// can't drift it towards now().
 	const past = () => new Date(Date.now() - 60_000);
 
+	let dataSource: DataSource;
 	let jobRepository: ScheduledJobRepository;
 	let taskRepository: ScheduledTaskRepository;
 	let job: ScheduledJobEntity;
@@ -56,7 +59,10 @@ describe('ScheduledTaskRepository executor methods', () => {
 	const reload = async (id: string) => await taskRepository.findOneByOrFail({ id });
 
 	beforeAll(async () => {
+		// Use separate connections for concurrent claims.
+		Container.get(GlobalConfig).database.postgresdb.poolSize = 4;
 		await testDb.init();
+		dataSource = Container.get(DataSource);
 		jobRepository = Container.get(ScheduledJobRepository);
 		taskRepository = Container.get(ScheduledTaskRepository);
 	});
@@ -239,6 +245,377 @@ describe('ScheduledTaskRepository executor methods', () => {
 		});
 	});
 
+	describe('claimDueTasks under a concurrencyLimit', () => {
+		const createLimitedJob = async (concurrencyLimit: number | null) => {
+			const jobName = `limited-${Math.random().toString(36).slice(2)}`;
+			return await jobRepository.save(
+				jobRepository.create({
+					name: jobName,
+					...selfOwned(jobName),
+					taskType: TASK_TYPE,
+					payload: {},
+					kind: 'interval',
+					intervalSeconds: 60,
+					enabled: true,
+					nextRunAt: new Date('2026-01-01T00:00:00.000Z'),
+					maxAttempts: 1,
+					concurrencyLimit,
+				}),
+			);
+		};
+
+		const statusOf = async (id: string) => (await reload(id)).status;
+
+		const runningCountOf = async (jobId: number) =>
+			await taskRepository.countBy({ jobId, status: 'running' });
+
+		it('claims one of two due occurrences of a limit-1 job and keeps the other pending', async () => {
+			const limited = await createLimitedJob(1);
+			const first = await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 30_000) });
+			const second = await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 20_000) });
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(claimed.map((t) => t.id)).toEqual([first.id]);
+			expect(await statusOf(second.id)).toBe('pending');
+			expect((await reload(second.id)).claimedBy).toBeNull();
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('counts an occurrence already running elsewhere against the limit, even past its deadline', async () => {
+			const limited = await createLimitedJob(1);
+			await createTask({
+				jobId: limited.id,
+				status: 'running',
+				claimedBy: HOST_B,
+				leaseExpiresAt: new Date(Date.now() + 60_000),
+				leaseEpoch: 1,
+				missedAfter: new Date(Date.now() - 60_000),
+			});
+			const due = await createTask({ jobId: limited.id });
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(claimed).toHaveLength(0);
+			expect(await statusOf(due.id)).toBe('pending');
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('never has two running occurrences of a limit-1 job across passes', async () => {
+			const limited = await createLimitedJob(1);
+			await createTask({ jobId: limited.id });
+			await createTask({ jobId: limited.id });
+
+			const first = await taskRepository.claimDueTasks(claimOpts());
+			const second = await taskRepository.claimDueTasks(claimOpts({ host: HOST_B }));
+
+			expect(first).toHaveLength(1);
+			expect(second).toHaveLength(0);
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('runs two of three due occurrences of a limit-2 job and keeps the third pending', async () => {
+			const limited = await createLimitedJob(2);
+			const a = await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 30_000) });
+			const b = await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 20_000) });
+			const c = await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 10_000) });
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(new Set(claimed.map((t) => t.id))).toEqual(new Set([a.id, b.id]));
+			expect(await statusOf(c.id)).toBe('pending');
+			expect(await runningCountOf(limited.id)).toBe(2);
+		});
+
+		it('claims a held occurrence once a slot frees up, while its deadline is ahead', async () => {
+			const limited = await createLimitedJob(1);
+			await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 30_000) });
+			const held = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 20_000),
+				missedAfter: new Date(Date.now() + 60_000),
+			});
+
+			const [running] = await taskRepository.claimDueTasks(claimOpts());
+			expect(await taskRepository.claimDueTasks(claimOpts())).toHaveLength(0);
+
+			await taskRepository.completeTask({
+				host: HOST_A,
+				id: running.id,
+				claimedEpoch: running.leaseEpoch,
+			});
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(claimed.map((t) => t.id)).toEqual([held.id]);
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('retires a held occurrence as missed once its deadline passes, without running it', async () => {
+			const limited = await createLimitedJob(1);
+			await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 30_000) });
+			const held = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 20_000),
+				missedAfter: new Date(Date.now() + 60_000),
+			});
+
+			expect(await taskRepository.claimDueTasks(claimOpts())).toHaveLength(1);
+			expect(await statusOf(held.id)).toBe('pending');
+
+			await taskRepository.update(held.id, { missedAfter: new Date(Date.now() - 1_000) });
+
+			expect(await taskRepository.claimDueTasks(claimOpts())).toHaveLength(0);
+			expect(await taskRepository.retireMissedPending(10)).toBe(1);
+
+			const retired = await reload(held.id);
+			expect(retired.status).toBe('missed');
+			expect(retired.claimedBy).toBeNull();
+			expect(retired.startedAt).toBeNull();
+		});
+
+		it('claims the next occurrence without waiting for an expired one to retire', async () => {
+			const limited = await createLimitedJob(1);
+			const expired = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 30_000),
+				missedAfter: new Date(Date.now() - 60_000),
+			});
+			const ready = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 20_000),
+			});
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(claimed.map((task) => task.id)).toEqual([ready.id]);
+			expect(await statusOf(expired.id)).toBe('pending');
+			expect(await taskRepository.retireMissedPending(10)).toBe(1);
+			expect(await statusOf(expired.id)).toBe('missed');
+		});
+
+		it('lets a retry past its deadline take a slot', async () => {
+			const limited = await createLimitedJob(1);
+			const retry = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 30_000),
+				missedAfter: new Date(Date.now() - 60_000),
+				attempts: 1,
+				maxAttempts: 2,
+			});
+			const next = await createTask({
+				jobId: limited.id,
+				runAt: new Date(Date.now() - 20_000),
+			});
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(claimed.map((task) => task.id)).toEqual([retry.id]);
+			expect(await statusOf(next.id)).toBe('pending');
+		});
+
+		it('fills the batch with other jobs when a limited job holds occurrences back', async () => {
+			const limited = await createLimitedJob(1);
+			for (const offset of [50_000, 40_000, 30_000]) {
+				await createTask({ jobId: limited.id, runAt: new Date(Date.now() - offset) });
+			}
+			// Later than every limited task, so a cap applied after the batch size would skip them.
+			const unlimitedA = await createTask({ runAt: new Date(Date.now() - 20_000) });
+			const unlimitedB = await createTask({ runAt: new Date(Date.now() - 10_000) });
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts({ batchSize: 3 }));
+
+			expect(claimed).toHaveLength(3);
+			expect(claimed.map((t) => t.id)).toEqual(
+				expect.arrayContaining([unlimitedA.id, unlimitedB.id]),
+			);
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('claims every due occurrence of an unlimited job', async () => {
+			const unlimited = await createLimitedJob(null);
+			const created = await Promise.all([
+				createTask({ jobId: unlimited.id }),
+				createTask({ jobId: unlimited.id }),
+				createTask({ jobId: unlimited.id }),
+			]);
+
+			const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+			expect(new Set(claimed.map((t) => t.id))).toEqual(new Set(created.map((t) => t.id)));
+		});
+
+		it('never lets two concurrent claimers exceed a limit together', async () => {
+			const limited = await createLimitedJob(1);
+			await Promise.all([
+				createTask({ jobId: limited.id }),
+				createTask({ jobId: limited.id }),
+				createTask({ jobId: limited.id }),
+			]);
+
+			const [a, b] = await Promise.all([
+				taskRepository.claimDueTasks(claimOpts({ host: HOST_A })),
+				taskRepository.claimDueTasks(claimOpts({ host: HOST_B })),
+			]);
+
+			expect(a.length + b.length).toBe(1);
+			expect(await runningCountOf(limited.id)).toBe(1);
+		});
+
+		it('keeps claiming unlimited work while another claimer holds a limited job', async () => {
+			const limited = await createLimitedJob(1);
+			await createTask({ jobId: limited.id });
+			const unlimited = await Promise.all([createTask(), createTask(), createTask(), createTask()]);
+
+			const [a, b] = await Promise.all([
+				taskRepository.claimDueTasks(claimOpts({ host: HOST_A })),
+				taskRepository.claimDueTasks(claimOpts({ host: HOST_B })),
+			]);
+
+			const ids = [...a, ...b].map((t) => t.id);
+			expect(new Set(ids).size).toBe(ids.length);
+			expect(ids).toEqual(expect.arrayContaining(unlimited.map((t) => t.id)));
+			expect(ids).toHaveLength(unlimited.length + 1);
+		});
+
+		// Only Postgres needs these checks. SQLite runs one claim at a time.
+		describe.runIf(isPostgres)('across concurrent Postgres claimers', () => {
+			const PAUSE_UPDATE = 46202026;
+			const PAUSE_SNAPSHOT = 46202027;
+
+			const jobTable = () => dataSource.driver.escape(jobRepository.metadata.tableName);
+			const taskTable = () => dataSource.driver.escape(taskRepository.metadata.tableName);
+
+			/** Settles once exactly one session is waiting on the given advisory lock. */
+			const waitForBlockedOn = async (key: number) =>
+				await expect
+					.poll(
+						async () => {
+							const rows = await dataSource.query<Array<{ count: string }>>(
+								`SELECT count(*) FROM pg_locks
+								  WHERE locktype = 'advisory' AND objid = $1 AND NOT granted`,
+								[key],
+							);
+							return Number(rows[0].count);
+						},
+						{ timeout: 5_000 },
+					)
+					.toBe(1);
+
+			it('fills the batch from other jobs when a limited job is locked', async () => {
+				const limited = await createLimitedJob(1);
+				// Earlier than the other task, so a claim that waited for the lock would return it first.
+				await createTask({ jobId: limited.id, runAt: new Date(Date.now() - 50_000) });
+				const available = await createTask({ runAt: new Date(Date.now() - 10_000) });
+				const locker = dataSource.createQueryRunner();
+				await locker.connect();
+				await locker.startTransaction();
+
+				try {
+					await locker.query(`SELECT "id" FROM ${jobTable()} WHERE "id" = $1 FOR NO KEY UPDATE`, [
+						limited.id,
+					]);
+
+					const claimed = await taskRepository.claimDueTasks(claimOpts());
+
+					expect(claimed.map((task) => task.id)).toEqual([available.id]);
+					expect(await runningCountOf(limited.id)).toBe(0);
+				} finally {
+					await locker.rollbackTransaction();
+					await locker.release();
+				}
+			});
+
+			it('claims nothing on a snapshot taken before another claimer bumped the job', async () => {
+				const limited = await createLimitedJob(1);
+				const first = await createTask({
+					jobId: limited.id,
+					runAt: new Date(Date.now() - 30_000),
+				});
+
+				// Pause claim A inside its update, after it picks its tasks.
+				await dataSource.query(`CREATE FUNCTION pause_claim_update() RETURNS trigger LANGUAGE plpgsql AS $$
+						BEGIN
+							IF NEW."claimedBy" = '${HOST_A}' THEN PERFORM pg_advisory_xact_lock(${PAUSE_UPDATE}); END IF;
+							RETURN NEW;
+						END $$`);
+				await dataSource.query(`CREATE TRIGGER pause_claim_update BEFORE UPDATE ON ${taskTable()}
+						FOR EACH ROW EXECUTE FUNCTION pause_claim_update()`);
+				// Pause claim B after it reads the tasks, before it reaches the job row.
+				await dataSource.query(`CREATE FUNCTION pause_claim_snapshot(types text[]) RETURNS text[] LANGUAGE plpgsql AS $$
+						BEGIN
+							PERFORM pg_advisory_xact_lock(${PAUSE_SNAPSHOT});
+							RETURN types;
+						END $$`);
+
+				const barrier = dataSource.createQueryRunner();
+				await barrier.connect();
+				const originalQuery = taskRepository.query.bind(taskRepository);
+				const querySpy = vi.spyOn(taskRepository, 'query');
+				let claimA: Promise<ScheduledTask[]> | undefined;
+				let claimB: Promise<ScheduledTask[]> | undefined;
+				let a: ScheduledTask[] = [];
+				let b: ScheduledTask[] = [];
+
+				try {
+					await barrier.query(`SELECT pg_advisory_lock(${PAUSE_UPDATE})`);
+					await barrier.query(`SELECT pg_advisory_lock(${PAUSE_SNAPSHOT})`);
+
+					claimA = taskRepository.claimDueTasks(claimOpts({ host: HOST_A }));
+					await waitForBlockedOn(PAUSE_UPDATE);
+
+					// An earlier task takes the job's only place, so B's old read would
+					// pick a task that A never claimed.
+					const earlier = new Date(first.runAt.getTime() - 1_000);
+					await dataSource.transaction(
+						async (manager) =>
+							await taskRepository.insertIgnoringDuplicates(manager, [
+								{
+									jobId: limited.id,
+									taskType: TASK_TYPE,
+									payload: {},
+									scheduledFor: earlier,
+									runAt: earlier,
+									maxAttempts: 1,
+									missedAfter: null,
+								},
+							]),
+					);
+
+					querySpy.mockImplementation(
+						async (sql: string, parameters?: unknown[]) =>
+							await originalQuery(
+								sql.replace('= ANY($3)', '= ANY(pause_claim_snapshot($3::text[]))'),
+								parameters,
+							),
+					);
+					claimB = taskRepository.claimDueTasks(claimOpts({ host: HOST_B }));
+					await waitForBlockedOn(PAUSE_SNAPSHOT);
+
+					await barrier.query(`SELECT pg_advisory_unlock(${PAUSE_UPDATE})`);
+					a = await claimA;
+					await barrier.query(`SELECT pg_advisory_unlock(${PAUSE_SNAPSHOT})`);
+					b = await claimB;
+				} finally {
+					await barrier.query('SELECT pg_advisory_unlock_all()');
+					if (claimA) a = await claimA;
+					if (claimB) b = await claimB;
+					querySpy.mockRestore();
+					await barrier.release();
+					await dataSource.query(`DROP TRIGGER pause_claim_update ON ${taskTable()}`);
+					await dataSource.query('DROP FUNCTION pause_claim_update()');
+					await dataSource.query('DROP FUNCTION pause_claim_snapshot(text[])');
+				}
+
+				expect(a.map((task) => task.id)).toEqual([first.id]);
+				expect(b).toHaveLength(0);
+				expect(await runningCountOf(limited.id)).toBe(1);
+				// The claim must not change the job's data.
+				expect(await jobRepository.findOneByOrFail({ id: limited.id })).toEqual(limited);
+			}, 20_000);
+		});
+	});
+
 	describe('guarded transitions', () => {
 		// Returns the id plus the epoch the claim set (the fencing token a terminal
 		// transition must present). First claim of a fresh row yields epoch 1.

packages/cli/test/integration/scheduling/scheduled-repositories.test.ts

@@ -1130,6 +1130,21 @@ describe('scheduled repositories', () => {
 			expect(snapshot.oldestPendingAgeMs).toBeNull();
 		});
 
+		it('counts a task held back by its concurrencyLimit as due', async () => {
+			const job = await createJob({ concurrencyLimit: 1 });
+			await createTask(job.id, {
+				status: 'running',
+				claimedBy: 'main-1',
+				leaseExpiresAt: secondsFromNow(60),
+			});
+			await createTask(job.id, { status: 'pending', runAt: secondsFromNow(-30) });
+
+			const snapshot = await taskRepository.getMetricSnapshot();
+
+			expect(snapshot.due).toBe(1);
+			expect(snapshot.oldestPendingAgeMs).not.toBeNull();
+		});
+
 		it('reports all-zero counts and a null age on an empty queue', async () => {
 			const snapshot = await taskRepository.getMetricSnapshot();
 

packages/cli/test/integration/scheduling/scheduler-execute.test.ts

@@ -1,5 +1,5 @@
 import { testDb } from '@n8n/backend-test-utils';
-import type { ScheduledJob } from '@n8n/db';
+import type { ScheduledJob, ScheduledTask } from '@n8n/db';
 import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
 import { createScheduler } from '@n8n/scheduler';
@@ -7,7 +7,7 @@ import type { ClaimedTask, Scheduler, SchedulerPasses } from '@n8n/scheduler';
 
 import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
 
-import { selfOwned } from './shared/job-factory';
+import { seedDueTask, selfOwned } from './shared/job-factory';
 
 /**
  * The composed path against a real database: the storage bindings
@@ -24,6 +24,8 @@ describe('scheduler execution over the storage bindings', () => {
 	let taskRepo: ScheduledTaskRepository;
 	let scheduler: Scheduler & SchedulerPasses;
 	const executed: ClaimedTask[] = [];
+	// When set, the handler waits for it, so a test can keep a task `running`.
+	let holdHandler: Promise<void> | null = null;
 
 	beforeAll(async () => {
 		await testDb.init();
@@ -41,6 +43,9 @@ describe('scheduler execution over the storage bindings', () => {
 		scheduler.registerTaskHandler(TASK_TYPE, {
 			execute: async (task, report) => {
 				executed.push(task);
+				if (holdHandler !== null) {
+					await holdHandler;
+				}
 				// The fake's effect is the push above; report it so the task carries its
 				// effect marker (`dispatchedAt`) like a real handler would.
 				return report.dispatched();
@@ -172,6 +177,37 @@ describe('scheduler execution over the storage bindings', () => {
 		expect(failed.errorMessage).toBe('Lease expired before completion');
 	});
 
+	it('runs at most concurrencyLimit occurrences of a job at once, then the held one', async () => {
+		const job = await createJob({ concurrencyLimit: 1 });
+		await seedDueTask(taskRepo, TASK_TYPE, job.id, 0);
+		await seedDueTask(taskRepo, TASK_TYPE, job.id, 1);
+		const countWithStatus = async (status: ScheduledTask['status']) =>
+			await taskRepo.countBy({ jobId: job.id, status });
+
+		let release: () => void = () => {};
+		holdHandler = new Promise<void>((resolve) => {
+			release = resolve;
+		});
+		try {
+			expect(await scheduler.execute()).toHaveLength(1);
+			await waitFor(async () => executed.length === 1);
+
+			// The second task stays pending while the first runs.
+			expect(await scheduler.execute()).toHaveLength(0);
+			expect(await countWithStatus('running')).toBe(1);
+			expect(await countWithStatus('pending')).toBe(1);
+		} finally {
+			release();
+			holdHandler = null;
+		}
+		await waitFor(async () => (await countWithStatus('succeeded')) === 1);
+
+		// Once the first task finishes, the second one runs.
+		expect(await scheduler.execute()).toHaveLength(1);
+		await waitFor(async () => (await countWithStatus('succeeded')) === 2);
+		expect(executed).toHaveLength(2);
+	}, 15_000);
+
 	// Last on purpose: it stops the shared scheduler's executor.
 	it('releases claimed but unfired tasks on stop', async () => {
 		const job = await createJob();

packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts

@@ -1,6 +1,7 @@
 import { ScheduledJobMisfirePolicy } from '@n8n/constants';
 import { testDb } from '@n8n/backend-test-utils';
 import {
+	dbNowLiteral,
 	ScheduledJob,
 	ScheduledJobRepository,
 	ScheduledTask,
@@ -53,6 +54,9 @@ const READ_ITERS = envInt('N8N_SCHEDULER_QUERY_ITERS', 50);
 // enough to skew later iterations.
 const WRITE_ITERS = envInt('N8N_SCHEDULER_QUERY_WRITE_ITERS', 25);
 const BATCH = envInt('N8N_SCHEDULER_QUERY_BATCH', 100);
+// One job in this many has a concurrencyLimit of 1 and a due task. Every other one
+// also has a running task that uses its place.
+const LIMITED_JOB_EVERY = envInt('N8N_SCHEDULER_QUERY_LIMITED_JOB_EVERY', 10);
 const WRITE_BATCH = envInt('N8N_SCHEDULER_QUERY_WRITE_BATCH', 10_000);
 
 // `insertMany` chunks its insert and `name IN (...)` read-back internally (dialect-aware,
@@ -166,11 +170,52 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => {
 				enabled,
 				nextRunAt: enabled ? (due ? secondsAgo(60) : secondsFromNow(3600)) : null,
 				maxAttempts: 1,
+				concurrencyLimit: i % LIMITED_JOB_EVERY === 0 ? 1 : null,
 			});
 		}
 		await bulkInsert(ScheduledJob, rows);
 	}
 
+	async function seedLimitedTasks(): Promise<{ limitedJobs: number; busyJobs: number }> {
+		const limitedJobs = await dataSource
+			.createQueryBuilder(ScheduledJob, 'j')
+			.select('j.id')
+			.where('j.concurrencyLimit IS NOT NULL')
+			.orderBy('j.id', 'ASC')
+			.getMany();
+		const rows: Array<QueryDeepPartialEntity<ScheduledTask>> = [];
+		let busyJobs = 0;
+		for (const [i, job] of limitedJobs.entries()) {
+			const due = secondsAgo(60 + i);
+			rows.push({
+				jobId: job.id,
+				taskType: TASK_TYPE,
+				payload: {},
+				scheduledFor: due,
+				runAt: due,
+				status: 'pending',
+				maxAttempts: 1,
+			});
+			if (i % 2 === 0) {
+				busyJobs += 1;
+				const started = secondsAgo(3600 + i);
+				rows.push({
+					jobId: job.id,
+					taskType: TASK_TYPE,
+					payload: {},
+					scheduledFor: started,
+					runAt: started,
+					status: 'running',
+					claimedBy: 'seed',
+					leaseExpiresAt: secondsFromNow(60),
+					maxAttempts: 1,
+				});
+			}
+		}
+		await bulkInsert(ScheduledTask, rows);
+		return { limitedJobs: limitedJobs.length, busyJobs };
+	}
+
 	beforeAll(async () => {
 		await testDb.init();
 		dataSource = Container.get(DataSource);
@@ -194,9 +239,13 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => {
 		);
 		await seedTasks(anchorJob.id);
 		await seedJobs();
+		const limited = await seedLimitedTasks();
+		// Refresh planner statistics so the plans match a live instance.
+		await dataSource.query('ANALYZE');
 		report('corpus seeded', {
-			'scheduled_task rows': commas(TASK_ROWS),
+			'scheduled_task rows': commas(TASK_ROWS + limited.limitedJobs + limited.busyJobs),
 			'scheduled_job rows': commas(JOB_ROWS),
+			'limited jobs (due / of which busy)': `${commas(limited.limitedJobs)} / ${commas(limited.busyJobs)}`,
 		});
 	}, TEST_TIMEOUT_MS);
 
@@ -305,7 +354,21 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => {
 			);
 
 			// ScheduledTaskRepository.claimDueTasks — the claim's candidate select
-			// (pending + due, ordered by runAt). Backed by the partial index on runAt.
+			// (pending + due, ordered by runAt, without limited jobs). Backed by the
+			// partial indexes on runAt and on limited jobs.
+			const claimCandidateSelect = dataSource
+				.createQueryBuilder(ScheduledTask, 't')
+				.where('t.status = :s', { s: 'pending' })
+				.andWhere('t.taskType IN (:...tt)', { tt: [TASK_TYPE] })
+				.andWhere('t.runAt <= :now', { now: dueParam })
+				.andWhere(
+					't.jobId NOT IN (SELECT "id" FROM ' +
+						dataSource.getRepository(ScheduledJob).metadata.tableName +
+						' WHERE "concurrencyLimit" IS NOT NULL)',
+				)
+				.orderBy('t.runAt', 'ASC')
+				.limit(BATCH)
+				.getQueryAndParameters();
 			await profileRead(
 				'ScheduledTask.claimDueTasks (candidate select)',
 				async () =>
@@ -316,22 +379,39 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => {
 						leaseMs: 60_000,
 						batchSize: BATCH,
 					}),
-				dataSource
-					.createQueryBuilder(ScheduledTask, 't')
-					.where('t.status = :s', { s: 'pending' })
-					.andWhere('t.taskType IN (:...tt)', { tt: [TASK_TYPE] })
-					.andWhere('t.runAt <= :now', { now: dueParam })
-					.orderBy('t.runAt', 'ASC')
-					.limit(BATCH)
-					.getQueryAndParameters()[0],
-				dataSource
-					.createQueryBuilder(ScheduledTask, 't')
-					.where('t.status = :s', { s: 'pending' })
-					.andWhere('t.taskType IN (:...tt)', { tt: [TASK_TYPE] })
-					.andWhere('t.runAt <= :now', { now: dueParam })
-					.orderBy('t.runAt', 'ASC')
-					.limit(BATCH)
-					.getQueryAndParameters()[1],
+				claimCandidateSelect[0],
+				claimCandidateSelect[1],
+			);
+
+			// The other half of the claim: which pending tasks of a limited job fit its
+			// limit. Its cost grows with the limited backlog. Keep in step with
+			// `ScheduledTaskRepository.allowedByConcurrencyLimitSql`.
+			const jobTable = dataSource.getMetadata(ScheduledJob).tablePath;
+			const limitedSlotSelect = `SELECT ranked."id"
+				FROM (
+					SELECT c."id", c."status",
+						ROW_NUMBER() OVER (
+							PARTITION BY c."jobId"
+							ORDER BY CASE WHEN c."status" = 'pending' THEN 0 ELSE 1 END, c."runAt", c."id"
+						) AS "slot",
+						cj."concurrencyLimit" - SUM(
+							CASE WHEN c."status" = 'running' THEN 1 ELSE 0 END
+						) OVER (PARTITION BY c."jobId") AS "freeSlots"
+					FROM ${taskTable} c
+					JOIN ${jobTable} cj ON cj."id" = c."jobId"
+					WHERE cj."concurrencyLimit" IS NOT NULL
+						AND (c."status" = 'running'
+							OR (c."status" = 'pending'
+								AND (c."missedAfter" IS NULL
+									OR c."missedAfter" > ${dbNowLiteral(isPostgres)}
+									OR c."attempts" > 0)))
+				) ranked
+				WHERE ranked."status" = 'pending' AND ranked."slot" <= ranked."freeSlots"`;
+			await profileRead(
+				'ScheduledTask.claimDueTasks (limited-job slot ranking)',
+				async () => await dataSource.query(limitedSlotSelect),
+				limitedSlotSelect,
+				[],
 			);
 
 			// ScheduledTaskRepository.findExpiredLeases — reaper sweep. Backed by the