From f6f71abef45660e7556670073dd5b901ccaaa0ed Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Fri, 2 Oct 2026 10:50:24 +0000 Subject: [PATCH] feat(core): Cap concurrent occurrences of a durable scheduled job (no-changelog) (#39197) --- .../repositories/scheduled-task.repository.ts | 120 ++++-- .../@n8n/scheduler/src/core/executor/store.ts | 3 + .../scheduling/executor-repository.test.ts | 379 +++++++++++++++++- .../scheduling/scheduled-repositories.test.ts | 15 + .../scheduling/scheduler-execute.test.ts | 40 +- .../scheduler-query-benchmarks.test.ts | 116 +++++- 6 files changed, 626 insertions(+), 47 deletions(-) diff --git a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts index 48b6cce103c..074b1b7da13 100644 --- a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts @@ -123,6 +123,7 @@ export type ScheduledTaskMetricSnapshot = { export class ScheduledTaskRepository extends Repository { 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 { 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 { * * 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 { 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 { } private async claimWithPostgres(opts: ClaimDueTasksOptions): Promise { + 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 { // 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 { * (`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. */ diff --git a/packages/@n8n/scheduler/src/core/executor/store.ts b/packages/@n8n/scheduler/src/core/executor/store.ts index 09bb9f76695..0b00292f366 100644 --- a/packages/@n8n/scheduler/src/core/executor/store.ts +++ b/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; diff --git a/packages/cli/test/integration/scheduling/executor-repository.test.ts b/packages/cli/test/integration/scheduling/executor-repository.test.ts index 3da2c6a57af..3967b494997 100644 --- a/packages/cli/test/integration/scheduling/executor-repository.test.ts +++ b/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>( + `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 | undefined; + let claimB: Promise | 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. diff --git a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts index 78c1e2b3d15..8f7a66ae2cc 100644 --- a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts +++ b/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(); diff --git a/packages/cli/test/integration/scheduling/scheduler-execute.test.ts b/packages/cli/test/integration/scheduling/scheduler-execute.test.ts index 5d7326799f7..9ec4007e334 100644 --- a/packages/cli/test/integration/scheduling/scheduler-execute.test.ts +++ b/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 | 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((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(); 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 89e8c740380..af07aa66528 100644 --- a/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts +++ b/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> = []; + 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