diff --git a/packages/@n8n/db/src/entities/scheduled-job.ts b/packages/@n8n/db/src/entities/scheduled-job.ts index 93e71968950..c41b2d846c8 100644 --- a/packages/@n8n/db/src/entities/scheduled-job.ts +++ b/packages/@n8n/db/src/entities/scheduled-job.ts @@ -158,6 +158,16 @@ export class ScheduledJob extends WithTimestamps { @Column({ type: 'int', default: 1 }) maxAttempts: number; + /** + * How long, in seconds, one attempt of an occurrence may run before the executor + * stops it. Copied onto each occurrence this job materializes. + * + * Pinned to a literal, not the config default, so a later change to the default + * can't retroactively change what this entity declares. + */ + @Column({ type: 'int', default: 300 }) + timeoutSeconds: number; + /** * How many of this job's occurrences may run at the same time. `null` means no * limit. diff --git a/packages/@n8n/db/src/entities/scheduled-task.ts b/packages/@n8n/db/src/entities/scheduled-task.ts index 1707357c062..4c1b22b7ad8 100644 --- a/packages/@n8n/db/src/entities/scheduled-task.ts +++ b/packages/@n8n/db/src/entities/scheduled-task.ts @@ -133,6 +133,14 @@ export class ScheduledTask extends WithCreatedAt { @Column({ type: 'int', default: 1 }) maxAttempts: number; + /** + * How long, in seconds, one attempt may run before the executor stops it. + * Copied from the job when this run is created, and updated while the run is + * `pending` when the job's timeout changes. + */ + @Column({ type: 'int', default: 300 }) + timeoutSeconds: number; + /** * Which worker is currently running this; * `null` when no one has claimed it. diff --git a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts index decd72ccefb..fb28eba35d0 100644 --- a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts @@ -28,6 +28,7 @@ const newJob = (name: string): NewScheduledJob => ({ fireAt: null, nextRunAt: CLOCK, maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: 'coalesce', misfireGraceSeconds: 60, concurrencyLimit: null, @@ -446,6 +447,7 @@ describe('ScheduledJobRepository', () => { fireAt: null, nextRunAt: CLOCK, maxAttempts: 3, + timeoutSeconds: 600, misfirePolicy: 'skip', misfireGraceSeconds: 120, concurrencyLimit: 2, diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts index b05a842e7b4..05f668533fc 100644 --- a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -35,6 +35,7 @@ export type NewScheduledJob = Pick< | 'fireAt' | 'nextRunAt' | 'maxAttempts' + | 'timeoutSeconds' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit' @@ -52,6 +53,7 @@ export type ScheduledJobDefinitionUpdate = Pick< | 'fireAt' | 'nextRunAt' | 'maxAttempts' + | 'timeoutSeconds' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit' @@ -290,7 +292,11 @@ export class ScheduledJobRepository extends Repository { ids: number[], update: Pick< ScheduledJob, - 'maxAttempts' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit' + | 'maxAttempts' + | 'timeoutSeconds' + | 'misfirePolicy' + | 'misfireGraceSeconds' + | 'concurrencyLimit' >, ): Promise { if (ids.length === 0) return; diff --git a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts index 3048668227f..b435e113a93 100644 --- a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts @@ -80,6 +80,7 @@ export interface NewOccurrence { scheduledFor: Date; runAt: Date; maxAttempts: number; + timeoutSeconds: number; missedAfter?: Date | null; } @@ -270,6 +271,21 @@ export class ScheduledTaskRepository extends Repository { return result.affected ?? 0; } + /** Set the timeout of the still-`pending` occurrences of the given jobs. */ + async updateTimeoutForPendingJobs( + manager: EntityManager, + jobIds: number[], + timeoutSeconds: number, + ): Promise { + if (jobIds.length > 0) { + await manager.update( + ScheduledTask, + { jobId: In(jobIds), status: ScheduledTaskStatus.Pending }, + { timeoutSeconds }, + ); + } + } + /** * Recompute `missedAfter` on a job's still-`pending` occurrences from the given * grace, anchored to each row's own `runAt` or DB-now, whichever is later, so a diff --git a/packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts b/packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts index ee4f09c3228..8380de713b3 100644 --- a/packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts +++ b/packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts @@ -46,6 +46,7 @@ describe('executor claims far enough ahead to fire on time', () => { status: 'pending', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 0, }); diff --git a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts b/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts index b41665f687e..67965ba352b 100644 --- a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts +++ b/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts @@ -250,6 +250,7 @@ const claimedTask = (overrides: Partial = {}): ClaimedTask => ({ status: ScheduledTaskStatus.Running, attempts: 0, maxAttempts: 3, + timeoutSeconds: 300, leaseEpoch: 1, ...overrides, }); @@ -355,6 +356,7 @@ describe('createScheduler materialize', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -402,6 +404,7 @@ describe('createScheduler materialize', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -1279,6 +1282,7 @@ describe('createScheduler tracing', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -1333,6 +1337,7 @@ describe('createScheduler tracing', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -1630,6 +1635,7 @@ describe('createScheduler metrics', () => { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 3, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/errors.ts b/packages/@n8n/scheduler/src/core/errors.ts index 2d66596b01b..566120683b3 100644 --- a/packages/@n8n/scheduler/src/core/errors.ts +++ b/packages/@n8n/scheduler/src/core/errors.ts @@ -70,6 +70,19 @@ export class LeaseLostError extends Error { } } +/** Raised when a job is provisioned with a run option outside the range the scheduler enforces. */ +export class InvalidRunOptionError extends Error { + constructor( + message: string, + readonly option: 'timeoutSeconds' | 'concurrencyLimit', + readonly value: number, + readonly max: number, + ) { + super(message); + this.name = 'InvalidRunOptionError'; + } +} + /** Raised when a lease heartbeat is given a lease duration that is not a positive integer. */ export class InvalidLeaseDurationError extends Error { constructor(readonly leaseDurationMs: number) { diff --git a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.property.test.ts b/packages/@n8n/scheduler/src/core/executor/__tests__/executor.property.test.ts index 863b1c10151..1c66b34cc03 100644 --- a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.property.test.ts +++ b/packages/@n8n/scheduler/src/core/executor/__tests__/executor.property.test.ts @@ -20,6 +20,7 @@ const claimedTask = (id: string): ClaimedTask => ({ status: 'running', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 1, }); diff --git a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts b/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts index 8840d1be6df..364b507566f 100644 --- a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts +++ b/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts @@ -23,6 +23,7 @@ const claimedTask = (overrides: Partial = {}): ClaimedTask => ({ status: 'running', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 1, ...overrides, }); diff --git a/packages/@n8n/scheduler/src/core/index.ts b/packages/@n8n/scheduler/src/core/index.ts index 3caf8d4cf90..cdffce9ff43 100644 --- a/packages/@n8n/scheduler/src/core/index.ts +++ b/packages/@n8n/scheduler/src/core/index.ts @@ -27,8 +27,21 @@ export { } from './clock-skew'; export type { ClockSkew, ClockSkewOptions } from './clock-skew'; -export { provision, deprovision, createJobProvisioner, scheduleFingerprint } from './provisioning'; +export { + provision, + deprovision, + createJobProvisioner, + scheduleFingerprint, + findOutdatedJobs, + resolveCoreRunOptions, +} from './provisioning'; export type { + MisfireGraceAdjustment, + OutdatedJobIds, + RequestedRunOptions, + ResolvedRunOptions, + RunOptionDefaults, + RunOptions, JobProvisioner, JobProvisionerDeps, OwnedScope, @@ -97,6 +110,7 @@ export { UnregisteredOwnerTypeError, InvalidOwnerIdError, InvalidOwnerMemberIdError, + InvalidRunOptionError, LeaseLostError, } from './errors'; export type { RetentionOptions, RetentionSummary } from './retention'; diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts index d912cc12aeb..4cf260186ca 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/coalesce-group.test.ts @@ -23,6 +23,7 @@ const makeJob = (overrides: Partial = {}): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.CoalesceOwner, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts index 1674d53a06c..638840aa105 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts @@ -22,6 +22,7 @@ const makeJob = (id: number): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -104,6 +105,7 @@ describe('materialize', () => { scheduledFor: NOW, runAt: NOW, maxAttempts: 1, + timeoutSeconds: 300, missedAfter: new Date('2026-01-01T00:01:00.000Z'), }); expect(tx.recordOccurrences).toHaveBeenCalledWith([occurrence(1), occurrence(2)]); diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts index 4fa23c9d687..9037a65ef9a 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.property.test.ts @@ -21,6 +21,7 @@ const makeIntervalJob = (intervalSeconds: number): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -41,6 +42,7 @@ const makeCronJob = (cronExpression: string, timezone: string): ScheduledJob => nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, @@ -63,6 +65,7 @@ const makeOneOffJob = (fireAt: Date): ScheduledJob => ({ nextRunAt: fireAt, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts index e6e2a477004..319f20f522c 100644 --- a/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts +++ b/packages/@n8n/scheduler/src/core/materializer/__tests__/plan.test.ts @@ -23,6 +23,7 @@ const makeJob = (overrides: Partial = {}): ScheduledJob => ({ nextRunAt: NOW, lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/materializer/materialize.ts b/packages/@n8n/scheduler/src/core/materializer/materialize.ts index 6bcb35ec2bb..1eb0bcfa1ef 100644 --- a/packages/@n8n/scheduler/src/core/materializer/materialize.ts +++ b/packages/@n8n/scheduler/src/core/materializer/materialize.ts @@ -165,6 +165,7 @@ function toNewOccurrences(planned: PlannedJob[], now: Date): NewOccurrence[] { scheduledFor: when, runAt, maxAttempts: job.maxAttempts, + timeoutSeconds: job.timeoutSeconds, missedAfter: new Date(visibleAt + job.misfireGraceSeconds * Time.seconds.toMilliseconds), }; }), diff --git a/packages/@n8n/scheduler/src/core/materializer/transaction.ts b/packages/@n8n/scheduler/src/core/materializer/transaction.ts index 16d68bece72..0f0c51070d5 100644 --- a/packages/@n8n/scheduler/src/core/materializer/transaction.ts +++ b/packages/@n8n/scheduler/src/core/materializer/transaction.ts @@ -13,6 +13,7 @@ export interface NewOccurrence { scheduledFor: Date; runAt: Date; maxAttempts: number; + timeoutSeconds: number; /** * When this occurrence stops being worth running, resolved here so the claim can * compare one column against the clock instead of joining the schedule and diff --git a/packages/@n8n/scheduler/src/core/provisioning/__tests__/provision.test.ts b/packages/@n8n/scheduler/src/core/provisioning/__tests__/provision.test.ts index b02f1835e73..20a0091a0eb 100644 --- a/packages/@n8n/scheduler/src/core/provisioning/__tests__/provision.test.ts +++ b/packages/@n8n/scheduler/src/core/provisioning/__tests__/provision.test.ts @@ -156,6 +156,7 @@ describe('provision seeding first occurrences', () => { nextRunAt: new Date(NOW.getTime() + 10_000), lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/provisioning/__tests__/provisioner.test.ts b/packages/@n8n/scheduler/src/core/provisioning/__tests__/provisioner.test.ts index ebe3d6eccbb..1b1c3f31d60 100644 --- a/packages/@n8n/scheduler/src/core/provisioning/__tests__/provisioner.test.ts +++ b/packages/@n8n/scheduler/src/core/provisioning/__tests__/provisioner.test.ts @@ -99,6 +99,7 @@ describe('createJobProvisioner', () => { nextRunAt: new Date(now.getTime() + 10_000), lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/provisioning/__tests__/run-options.test.ts b/packages/@n8n/scheduler/src/core/provisioning/__tests__/run-options.test.ts new file mode 100644 index 00000000000..737cd9f042c --- /dev/null +++ b/packages/@n8n/scheduler/src/core/provisioning/__tests__/run-options.test.ts @@ -0,0 +1,282 @@ +import { ScheduledJobMisfirePolicy } from '@n8n/constants'; + +import { InvalidRunOptionError } from '../../errors'; +import { + findOutdatedJobs, + resolveCoreRunOptions, + type RequestedRunOptions, + type RunOptionDefaults, + type RunOptions, +} from '../run-options'; + +const THIRTY_DAYS_IN_SECONDS = 30 * 24 * 60 * 60; + +const DEFAULTS: RunOptionDefaults = { + maxAttempts: 5, + timeoutSeconds: 300, + misfireGraceSeconds: 90, + executorIntervalSeconds: 5, + materializationWindowSeconds: 60, +}; + +const resolve = ( + requested: Partial = {}, + defaults: Partial = {}, +) => + resolveCoreRunOptions( + { misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, ...requested }, + { ...DEFAULTS, ...defaults }, + ); + +const resolveGrace = (misfireGraceSeconds: unknown, defaults: Partial = {}) => + resolve({ misfireGraceSeconds }, defaults); + +describe('resolveCoreRunOptions', () => { + it('fills every omitted option from the instance defaults', () => { + expect(resolve()).toEqual({ + runOptions: { + maxAttempts: 5, + timeoutSeconds: 300, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + misfireGraceSeconds: 90, + concurrencyLimit: null, + }, + }); + }); + + describe('misfire grace', () => { + it('resolves a requested grace in place of the configured grace', () => { + expect(resolveGrace(300).runOptions.misfireGraceSeconds).toBe(300); + }); + + it('raises a requested grace equal to the executor interval to one second above the interval', () => { + const { runOptions } = resolveGrace(120, { + executorIntervalSeconds: 120, + materializationWindowSeconds: 60, + }); + + expect(runOptions.misfireGraceSeconds).toBe(121); + }); + + // A grace the node did not really supply falls back to the instance value, not + // to a floor: below-one values (including `null`, which coerces to zero) and + // values that are not a finite number at all. + it.each([ + { name: 'zero', grace: 0 }, + { name: 'null', grace: null }, + { name: 'a negative value', grace: -5 }, + { name: 'a fraction below one', grace: 0.5 }, + { name: 'NaN', grace: Number.NaN }, + { name: 'undefined', grace: undefined }, + { name: 'Infinity', grace: Number.POSITIVE_INFINITY }, + { name: 'a non-numeric string', grace: 'not-a-number' }, + ])('resolves $name to the instance-configured grace rather than to a floor', ({ grace }) => { + expect(resolveGrace(grace)).toEqual({ + runOptions: expect.objectContaining({ misfireGraceSeconds: 90 }), + }); + }); + + it('truncates a fractional requested grace to whole seconds', () => { + expect(resolveGrace(300.5).runOptions.misfireGraceSeconds).toBe(300); + }); + + it('leaves an instance-configured grace below the floors unclamped', () => { + expect(resolveGrace(undefined, { misfireGraceSeconds: 10 }).runOptions).toEqual( + expect.objectContaining({ misfireGraceSeconds: 10 }), + ); + }); + + it('resolves a requested grace given as a numeric string to that number', () => { + expect(resolveGrace('300').runOptions.misfireGraceSeconds).toBe(300); + }); + + it('accepts a requested grace of one second, raising it to the floor', () => { + expect(resolveGrace(1).runOptions.misfireGraceSeconds).toBe(60); + }); + + it('leaves a requested grace sitting exactly on the floor unadjusted', () => { + expect(resolveGrace(60)).toEqual({ + runOptions: expect.objectContaining({ misfireGraceSeconds: 60 }), + }); + }); + + it('leaves a requested grace sitting exactly on the thirty-day cap unadjusted', () => { + expect(resolveGrace(THIRTY_DAYS_IN_SECONDS)).toEqual({ + runOptions: expect.objectContaining({ misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS }), + }); + }); + + it('reports that it raised the grace, with the raw request', () => { + const { runOptions, misfireGraceAdjustment } = resolveGrace(30.7); + + expect(runOptions.misfireGraceSeconds).toBe(60); + expect(misfireGraceAdjustment).toEqual({ + direction: 'raised', + requestedMisfireGraceSeconds: 30.7, + }); + }); + + it('lowers a requested grace above the thirty-day cap to the cap, reporting that it lowered it', () => { + const { runOptions, misfireGraceAdjustment } = resolveGrace(THIRTY_DAYS_IN_SECONDS + 500); + + expect(runOptions.misfireGraceSeconds).toBe(THIRTY_DAYS_IN_SECONDS); + expect(misfireGraceAdjustment).toEqual({ + direction: 'lowered', + requestedMisfireGraceSeconds: THIRTY_DAYS_IN_SECONDS + 500, + }); + }); + + it('reports a fractional grace just above the thirty-day cap, whose truncation alone lands it on the cap', () => { + const { runOptions, misfireGraceAdjustment } = resolveGrace(THIRTY_DAYS_IN_SECONDS + 0.5); + + expect(runOptions.misfireGraceSeconds).toBe(THIRTY_DAYS_IN_SECONDS); + expect(misfireGraceAdjustment).toEqual({ + direction: 'lowered', + requestedMisfireGraceSeconds: THIRTY_DAYS_IN_SECONDS + 0.5, + }); + }); + + it('raises a requested grace to the thirty-day cap when the configured floors exceed the cap', () => { + const { runOptions, misfireGraceAdjustment } = resolveGrace(300, { + materializationWindowSeconds: THIRTY_DAYS_IN_SECONDS + 1000, + }); + + expect(runOptions.misfireGraceSeconds).toBe(THIRTY_DAYS_IN_SECONDS); + expect(misfireGraceAdjustment).toEqual({ + direction: 'raised', + requestedMisfireGraceSeconds: 300, + }); + }); + + it.each([ + { + name: 'the materialisation window', + defaults: { materializationWindowSeconds: undefined as unknown as number }, + }, + { + name: 'the executor interval', + defaults: { executorIntervalSeconds: undefined as unknown as number }, + }, + ])( + 'falls back to the instance-configured grace when $name is not configured', + ({ defaults }) => { + expect(resolveGrace(300, defaults).runOptions.misfireGraceSeconds).toBe(90); + }, + ); + }); + + describe('attempts', () => { + it('resolves a requested ceiling in place of the configured one', () => { + expect(resolve({ maxAttempts: 1 }).runOptions.maxAttempts).toBe(1); + }); + }); + + describe('timeout', () => { + it('resolves a requested timeout in place of the configured one', () => { + expect(resolve({ timeoutSeconds: 45 }).runOptions.timeoutSeconds).toBe(45); + }); + + it.each([0, -1, 1.5, 2_147_484])('rejects a timeout of %s', (timeoutSeconds) => { + expect(() => resolve({ timeoutSeconds })).toThrow(InvalidRunOptionError); + }); + + it('rejects a configured timeout outside the range', () => { + expect(() => resolve({}, { timeoutSeconds: 0 })).toThrow(InvalidRunOptionError); + }); + + it('accepts the longest timeout a timer honors', () => { + expect(resolve({ timeoutSeconds: 2_147_483 }).runOptions.timeoutSeconds).toBe(2_147_483); + }); + }); + + describe('concurrency limit', () => { + it('resolves a requested limit', () => { + expect(resolve({ concurrencyLimit: 2 }).runOptions.concurrencyLimit).toBe(2); + }); + + it.each([undefined, null])('resolves a limit of %s to no limit', (concurrencyLimit) => { + expect(resolve({ concurrencyLimit }).runOptions.concurrencyLimit).toBeNull(); + }); + + it('accepts the largest limit the column holds', () => { + expect(resolve({ concurrencyLimit: 2_147_483_647 }).runOptions.concurrencyLimit).toBe( + 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', + (concurrencyLimit) => { + expect(() => resolve({ concurrencyLimit })).toThrow(InvalidRunOptionError); + }, + ); + }); +}); + +describe('findOutdatedJobs', () => { + const RUN_OPTIONS: RunOptions = { + maxAttempts: 5, + timeoutSeconds: 300, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + misfireGraceSeconds: 90, + concurrencyLimit: null, + }; + const PAYLOAD = { workflowId: 'wf' }; + + const row = (over: Partial }> = {}) => ({ + id: 10, + payload: { workflowId: 'wf' }, + ...RUN_OPTIONS, + ...over, + }); + + const NONE = { runOptions: [], misfireGrace: [], timeout: [], payload: [] }; + + it('lists no job whose run options and payload match', () => { + expect(findOutdatedJobs([row()], RUN_OPTIONS, PAYLOAD)).toEqual(NONE); + }); + + it('lists a job whose grace changed for its run options and its queued deadlines', () => { + expect(findOutdatedJobs([row({ misfireGraceSeconds: 60 })], RUN_OPTIONS, PAYLOAD)).toEqual({ + ...NONE, + runOptions: [10], + misfireGrace: [10], + }); + }); + + it('lists a job whose timeout changed for its run options and its pending occurrences', () => { + expect(findOutdatedJobs([row({ timeoutSeconds: 600 })], RUN_OPTIONS, PAYLOAD)).toEqual({ + ...NONE, + runOptions: [10], + timeout: [10], + }); + }); + + it.each([ + { name: 'policy', over: { misfirePolicy: ScheduledJobMisfirePolicy.Skip } }, + { name: 'retry ceiling', over: { maxAttempts: 1 } }, + { name: 'concurrency limit', over: { concurrencyLimit: 2 } }, + ])('lists a job whose $name changed for its run options only', ({ over }) => { + expect(findOutdatedJobs([row(over)], RUN_OPTIONS, PAYLOAD)).toEqual({ + ...NONE, + runOptions: [10], + }); + }); + + it('lists a job whose grace and policy both changed once in the run options', () => { + const outdated = findOutdatedJobs( + [row({ misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 60 })], + RUN_OPTIONS, + PAYLOAD, + ); + + expect(outdated.runOptions).toEqual([10]); + }); + + it('lists a job whose stored payload differs by value', () => { + expect( + findOutdatedJobs([row({ payload: { workflowId: 'other' } })], RUN_OPTIONS, PAYLOAD), + ).toEqual({ ...NONE, payload: [10] }); + }); +}); diff --git a/packages/@n8n/scheduler/src/core/provisioning/index.ts b/packages/@n8n/scheduler/src/core/provisioning/index.ts index 38e2748ce53..956aa2d393c 100644 --- a/packages/@n8n/scheduler/src/core/provisioning/index.ts +++ b/packages/@n8n/scheduler/src/core/provisioning/index.ts @@ -1,6 +1,15 @@ export { provision, deprovision } from './provision'; export { scheduleFingerprint } from './schedule-identity'; export { createJobProvisioner } from './provisioner'; +export { findOutdatedJobs, resolveCoreRunOptions } from './run-options'; +export type { + MisfireGraceAdjustment, + OutdatedJobIds, + RequestedRunOptions, + ResolvedRunOptions, + RunOptionDefaults, + RunOptions, +} from './run-options'; export type { JobProvisioner, JobProvisionerDeps, OwnedScope } from './provisioner'; export type { ProvisionTransaction, diff --git a/packages/@n8n/scheduler/src/core/provisioning/run-options.ts b/packages/@n8n/scheduler/src/core/provisioning/run-options.ts new file mode 100644 index 00000000000..c61f7923a85 --- /dev/null +++ b/packages/@n8n/scheduler/src/core/provisioning/run-options.ts @@ -0,0 +1,203 @@ +import { + MAX_INTEGER_32BITS_SIGNED, + MAX_TIMER_DELAY_SECONDS, + type ScheduledJobMisfirePolicy, + Time, +} from '@n8n/constants'; +import { isDeepStrictEqual } from 'node:util'; + +import { InvalidRunOptionError } from '../errors'; +import type { ScheduledJob } from '../types/schedule'; + +/** + * Ceiling for a resolved misfire grace: the cap the config value carries, and well + * inside the column's `int` range. + */ +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; + +/** The settings every occurrence of a job runs with. */ +export type RunOptions = Pick< + ScheduledJob, + 'maxAttempts' | 'timeoutSeconds' | 'misfirePolicy' | 'misfireGraceSeconds' | 'concurrencyLimit' +>; + +/** Run options as a caller asked for them. An omitted value inherits the instance default. */ +export interface RequestedRunOptions { + misfirePolicy: ScheduledJobMisfirePolicy; + /** Not trusted to be a number: a stored node parameter can still be a string or `null`. */ + misfireGraceSeconds?: unknown; + maxAttempts?: number; + timeoutSeconds?: number; + concurrencyLimit?: number | null; +} + +/** The instance settings a request falls back to, and the bounds of the grace floor. */ +export interface RunOptionDefaults { + maxAttempts: number; + timeoutSeconds: number; + misfireGraceSeconds: number; + executorIntervalSeconds: number; + materializationWindowSeconds: number; +} + +/** A requested grace that resolution moved to the scheduler's floor or ceiling. */ +export interface MisfireGraceAdjustment { + direction: 'raised' | 'lowered'; + requestedMisfireGraceSeconds: number; +} + +export interface ResolvedRunOptions { + runOptions: RunOptions; + /** Set when the requested grace was moved to a bound. */ + misfireGraceAdjustment?: MisfireGraceAdjustment; +} + +/** The ids of stored jobs whose columns differ from the resolved request. */ +export interface OutdatedJobIds { + /** Any run option differs. */ + runOptions: number[]; + misfireGrace: number[]; + timeout: number[]; + payload: number[]; +} + +/** + * Resolve a request against the instance defaults into the run options a job stores. + * + * @throws {InvalidRunOptionError} when the timeout or the concurrency limit is outside + * the range the scheduler enforces. + */ +export function resolveCoreRunOptions( + requested: RequestedRunOptions, + defaults: RunOptionDefaults, +): ResolvedRunOptions { + const { misfireGraceSeconds, misfireGraceAdjustment } = resolveMisfireGraceSeconds( + requested.misfireGraceSeconds, + defaults, + ); + return { + runOptions: { + maxAttempts: requested.maxAttempts ?? defaults.maxAttempts, + timeoutSeconds: resolveTimeoutSeconds(requested.timeoutSeconds ?? defaults.timeoutSeconds), + misfirePolicy: requested.misfirePolicy, + misfireGraceSeconds, + concurrencyLimit: resolveConcurrencyLimit(requested.concurrencyLimit), + }, + misfireGraceAdjustment, + }; +} + +/** List the stored jobs whose run options or payload differ from the given ones. */ +export function findOutdatedJobs( + rows: Array & RunOptions>, + runOptions: RunOptions, + payload: Record, +): OutdatedJobIds { + const idsWhere = (differs: (row: (typeof rows)[number]) => boolean) => + rows.filter(differs).map((row) => row.id); + const graceChanged = (row: RunOptions) => + row.misfireGraceSeconds !== runOptions.misfireGraceSeconds; + const timeoutChanged = (row: RunOptions) => row.timeoutSeconds !== runOptions.timeoutSeconds; + return { + runOptions: idsWhere( + (row) => + graceChanged(row) || + timeoutChanged(row) || + row.misfirePolicy !== runOptions.misfirePolicy || + row.maxAttempts !== runOptions.maxAttempts || + row.concurrencyLimit !== runOptions.concurrencyLimit, + ), + misfireGrace: idsWhere(graceChanged), + timeout: idsWhere(timeoutChanged), + payload: idsWhere((row) => !isDeepStrictEqual(row.payload, payload)), + }; +} + +function resolveMisfireGraceSeconds( + requested: unknown, + defaults: RunOptionDefaults, +): { misfireGraceSeconds: number; misfireGraceAdjustment?: MisfireGraceAdjustment } { + const { misfireGraceSeconds, executorIntervalSeconds, materializationWindowSeconds } = defaults; + + const numeric = Number(requested); + if (!Number.isFinite(numeric)) { + return { misfireGraceSeconds }; + } + + const truncated = Math.trunc(numeric); + if (truncated < 1) { + return { misfireGraceSeconds }; + } + + const floor = Math.min( + Math.max(executorIntervalSeconds + 1, materializationWindowSeconds), + MAX_MISFIRE_GRACE_SECONDS, + ); + + if (!Number.isFinite(floor)) { + return { misfireGraceSeconds }; + } + + const effective = Math.min(Math.max(truncated, floor), MAX_MISFIRE_GRACE_SECONDS); + + if (effective === truncated && numeric <= MAX_MISFIRE_GRACE_SECONDS) { + return { misfireGraceSeconds: effective }; + } + + return { + misfireGraceSeconds: effective, + misfireGraceAdjustment: { + direction: effective > truncated ? 'raised' : 'lowered', + requestedMisfireGraceSeconds: numeric, + }, + }; +} + +/** + * Check a timeout against what the executor can enforce: a whole number of seconds + * from 1 to {@link MAX_TIMER_DELAY_SECONDS}. + * + * @throws {InvalidRunOptionError} when the timeout falls outside that range. A timeout + * of 0 stops every run as soon as it starts. + */ +function resolveTimeoutSeconds(requested: number): number { + if (!Number.isInteger(requested) || requested < 1 || requested > MAX_TIMER_DELAY_SECONDS) { + throw new InvalidRunOptionError( + 'Scheduled job timeout is outside the range the scheduler enforces', + 'timeoutSeconds', + requested, + MAX_TIMER_DELAY_SECONDS, + ); + } + return requested; +} + +/** + * 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 {InvalidRunOptionError} 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 InvalidRunOptionError( + 'Scheduled job concurrency limit is outside the range the column holds', + 'concurrencyLimit', + requested, + MAX_CONCURRENCY_LIMIT, + ); + } + return requested; +} diff --git a/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.property.test.ts b/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.property.test.ts index 6e70515eb40..0713f19239d 100644 --- a/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.property.test.ts +++ b/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.property.test.ts @@ -36,6 +36,7 @@ describe('reap decision (fast-check)', () => { scheduledFor: new Date('2026-01-01T00:00:00.000Z'), runAt: new Date('2026-01-01T00:00:00.000Z'), status: 'running', + timeoutSeconds: 300, dispatchedAt: dispatched ? new Date('2026-01-01T00:00:00.000Z') : null, ...row, }), diff --git a/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.test.ts b/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.test.ts index d6bbdd79a11..cbe14c720f6 100644 --- a/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.test.ts +++ b/packages/@n8n/scheduler/src/core/reaper/__tests__/reap.test.ts @@ -24,6 +24,7 @@ const expiredTask = (overrides: Partial = {}): ExpiredLeaseRow status: 'running', attempts: 0, maxAttempts: 3, + timeoutSeconds: 300, leaseEpoch: 1, // Pre-dispatch by default; post-dispatch tests set a concrete `dispatchedAt`. dispatchedAt: null, diff --git a/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts b/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts index 7eb2258bbab..10901569afc 100644 --- a/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts +++ b/packages/@n8n/scheduler/src/core/recurrence/__tests__/resolve.test.ts @@ -22,6 +22,7 @@ function makeJob(overrides: Partial = {}): ScheduledJob { nextRunAt: new Date('2026-01-01T00:00:00.000Z'), lastFiredAt: null, maxAttempts: 1, + timeoutSeconds: 300, concurrencyLimit: null, misfirePolicy: 'coalesce', misfireGraceSeconds: 60, diff --git a/packages/@n8n/scheduler/src/core/types/schedule.ts b/packages/@n8n/scheduler/src/core/types/schedule.ts index a3cc1b3bc15..36de0a9d2ee 100644 --- a/packages/@n8n/scheduler/src/core/types/schedule.ts +++ b/packages/@n8n/scheduler/src/core/types/schedule.ts @@ -103,6 +103,8 @@ export interface ScheduledJob { nextRunAt: Date | null; // the next instant the materializer materializes from. lastFiredAt: Date | null; maxAttempts: number; + /** How long one attempt of an occurrence may run before the executor stops it. */ + timeoutSeconds: number; /** * How many of this job's occurrences may run at the same time. `null` means no * limit. At least 1 when set. diff --git a/packages/@n8n/scheduler/src/core/types/task.ts b/packages/@n8n/scheduler/src/core/types/task.ts index 3d5cd55a895..2d117f14f68 100644 --- a/packages/@n8n/scheduler/src/core/types/task.ts +++ b/packages/@n8n/scheduler/src/core/types/task.ts @@ -39,6 +39,7 @@ export interface ScheduledTask { status: ScheduledTaskStatus; attempts: number; maxAttempts: number; + timeoutSeconds: number; } /** diff --git a/packages/@n8n/scheduler/src/observability/__tests__/attributes.test.ts b/packages/@n8n/scheduler/src/observability/__tests__/attributes.test.ts index f7b904fcef7..d99efe73147 100644 --- a/packages/@n8n/scheduler/src/observability/__tests__/attributes.test.ts +++ b/packages/@n8n/scheduler/src/observability/__tests__/attributes.test.ts @@ -20,6 +20,7 @@ const claimedTask: ClaimedTask = { status: 'running', attempts: 1, maxAttempts: 3, + timeoutSeconds: 300, leaseEpoch: 2, }; diff --git a/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts b/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts index 5e85d09b35a..2f33eb4a3d4 100644 --- a/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts +++ b/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts @@ -19,6 +19,7 @@ const claimedTask = (overrides: Partial = {}): ClaimedTask => ({ status: 'running', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 1, ...overrides, }); diff --git a/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts b/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts index c6eb31915be..1856d241351 100644 --- a/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts +++ b/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts @@ -34,6 +34,7 @@ describe('AgentTaskTaskHandler', () => { status: 'running', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 1, ...overrides, }); diff --git a/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts b/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts index c378be661b6..349c8aaf82d 100644 --- a/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts +++ b/packages/cli/src/scheduling/__tests__/durable-job-provisioner.test.ts @@ -53,6 +53,7 @@ const jobRow = ({ payload = {}, ...over }: Partial = {}): Schedule fireAt: null, nextRunAt: CLOCK, maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -84,6 +85,7 @@ describe('DurableJobProvisioner', () => { executorIntervalSeconds: 5, maxAttempts: 5, misfireGraceSeconds: 90, + taskTimeoutSeconds: 300, ...scheduler, }, generic: { timezone: 'UTC' }, @@ -142,6 +144,19 @@ describe('DurableJobProvisioner', () => { maxAttempts, }); + const provisionWithTimeout = async ( + timeoutSeconds: number | undefined, + desired: DesiredJob[] = [desiredJob('wf:node:0')], + ): Promise => + await provisioner.provision({ + owner: OWNER, + taskType: 'schedule-trigger', + payload: {}, + desired, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + timeoutSeconds, + }); + const provisionWithConcurrencyLimit = async ( concurrencyLimit: number | null | undefined, desired: DesiredJob[] = [desiredJob('wf:node:0')], @@ -203,6 +218,7 @@ describe('DurableJobProvisioner', () => { fireAt: null, nextRunAt: CLOCK, maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -271,6 +287,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -307,6 +324,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -367,6 +385,7 @@ describe('DurableJobProvisioner', () => { fireAt: null, nextRunAt: CLOCK, maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -467,81 +486,6 @@ describe('DurableJobProvisioner', () => { ); }); - it('raises a node-supplied grace equal to the executor interval to one second above the interval', async () => { - provisioner = makeProvisioner({ - executorIntervalSeconds: 120, - materializationWindowSeconds: 60, - }); - - await provisionWithGrace(120); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 121 }), - ]); - }); - - // A grace the node did not really supply falls back to the instance value, not - // to a floor: below-one values (including `null`, which coerces to zero) and - // values that are not a finite number at all. - it.each([ - { name: 'zero', grace: 0 }, - { name: 'null', grace: null }, - { name: 'a negative value', grace: -5 }, - { name: 'a fraction below one', grace: 0.5 }, - { name: 'NaN', grace: Number.NaN }, - { name: 'undefined', grace: undefined }, - { name: 'Infinity', grace: Number.POSITIVE_INFINITY }, - { name: 'a non-numeric string', grace: 'not-a-number' }, - ])( - 'resolves $name to the instance-configured grace rather than to a floor', - async ({ grace }) => { - await provisionWithGrace(grace); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 90 }), - ]); - }, - ); - - it('truncates a fractional node-supplied grace to whole seconds before writing it', async () => { - await provisionWithGrace(300.5); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 300 }), - ]); - }); - - it('leaves an instance-configured grace below the floors unclamped', async () => { - provisioner = makeProvisioner({ misfireGraceSeconds: 10 }); - - await provision( - 'schedule-trigger', - {}, - [desiredJob('wf:node:0')], - ScheduledJobMisfirePolicy.Coalesce, - ); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 10 }), - ]); - }); - - it('resolves a node-supplied grace given as a numeric string to that number', async () => { - await provisionWithGrace('300'); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 300 }), - ]); - }); - - it('accepts a node-supplied grace of one second, raising it to the floor', async () => { - await provisionWithGrace(1); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 60 }), - ]); - }); - it('leaves a node-supplied grace sitting exactly on the floor unclamped, and does not warn', async () => { await provisionWithGrace(60); @@ -551,15 +495,6 @@ describe('DurableJobProvisioner', () => { expect(logger.warn).not.toHaveBeenCalled(); }); - it('leaves a node-supplied grace sitting exactly on the thirty-day cap unclamped, and does not warn', async () => { - await provisionWithGrace(THIRTY_DAYS_IN_SECONDS); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS }), - ]); - expect(logger.warn).not.toHaveBeenCalled(); - }); - it('warns that it raised the grace, reporting the raw request, the value written and the owning node', async () => { await provisionWithGrace(30.7); @@ -589,64 +524,6 @@ describe('DurableJobProvisioner', () => { ); }); - it('warns about a fractional grace just above the thirty-day cap, whose truncation alone lands it on the cap', async () => { - await provisionWithGrace(THIRTY_DAYS_IN_SECONDS + 0.5); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS }), - ]); - expect(logger.warn).toHaveBeenCalledWith( - "Lowered a node's misfire grace to the scheduler's maximum", - { - ...OWNER, - requestedMisfireGraceSeconds: THIRTY_DAYS_IN_SECONDS + 0.5, - misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS, - }, - ); - }); - - it('raises a node-supplied grace to the thirty-day cap when the configured floors exceed the cap, and warns that it raised it', async () => { - provisioner = makeProvisioner({ - materializationWindowSeconds: THIRTY_DAYS_IN_SECONDS + 1000, - }); - - await provisionWithGrace(300); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS }), - ]); - expect(logger.warn).toHaveBeenCalledWith( - "Raised a node's misfire grace to the scheduler's minimum", - { - ...OWNER, - requestedMisfireGraceSeconds: 300, - misfireGraceSeconds: THIRTY_DAYS_IN_SECONDS, - }, - ); - }); - - it.each([ - { - name: 'the materialisation window', - config: { materializationWindowSeconds: undefined as unknown as number }, - }, - { - name: 'the executor interval', - config: { executorIntervalSeconds: undefined as unknown as number }, - }, - ])( - 'falls back to the instance-configured grace when $name is not configured', - async ({ config }) => { - provisioner = makeProvisioner(config); - - await provisionWithGrace(300); - - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ misfireGraceSeconds: 90 }), - ]); - }, - ); - it('treats a row already stored at the clamped grace as unchanged, leaving its queued tasks alone', async () => { jobs.findManyByOwner.mockResolvedValue([jobRow({ misfireGraceSeconds: 60 })]); @@ -662,6 +539,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, concurrencyLimit: null, @@ -678,6 +556,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 300, concurrencyLimit: null, @@ -728,6 +607,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 1, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, @@ -746,6 +626,74 @@ describe('DurableJobProvisioner', () => { }); }); + describe('timeout resolution', () => { + it('stamps a request-supplied timeout onto the inserted row, in place of the configured one', async () => { + await provisionWithTimeout(45); + + expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ + expect.objectContaining({ timeoutSeconds: 45 }), + ]); + }); + + it('stamps the configured timeout onto the inserted row when the request omits one', async () => { + provisioner = makeProvisioner({ taskTimeoutSeconds: 120 }); + + await provisionWithTimeout(undefined); + + expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ + expect.objectContaining({ timeoutSeconds: 120 }), + ]); + }); + + it("writes a request-supplied timeout onto a redefined job's row", async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow()]); + + await provisionWithTimeout(45, [ + desiredJob('wf:node:0', { + kind: 'cron', + cronExpression: '0 0 18 * * *', + timezone: 'UTC', + }), + ]); + + expect(jobs.updateDefinition).toHaveBeenCalledWith( + manager, + 10, + expect.objectContaining({ timeoutSeconds: 45 }), + ); + }); + + it('reconciles the timeout of a job and its pending occurrences when its schedule is unchanged', async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow({ timeoutSeconds: 300 })]); + + await provisionWithTimeout(600); + + expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { + maxAttempts: 5, + timeoutSeconds: 600, + misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, + misfireGraceSeconds: 90, + concurrencyLimit: null, + }); + expect(tasks.updateTimeoutForPendingJobs).toHaveBeenCalledWith(manager, [10], 600); + expect(jobs.updateDefinition).not.toHaveBeenCalled(); + }); + + it('leaves a job already stored at the requested timeout and its occurrences alone', async () => { + jobs.findManyByOwner.mockResolvedValue([jobRow({ timeoutSeconds: 600 })]); + + await provisionWithTimeout(600); + + expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [], expect.anything()); + expect(tasks.updateTimeoutForPendingJobs).toHaveBeenCalledWith(manager, [], 600); + }); + + it('rejects an out-of-range timeout with a user error', async () => { + await expect(provisionWithTimeout(0)).rejects.toThrow(UserError); + expect(jobs.insertMany).not.toHaveBeenCalled(); + }); + }); + describe('concurrency limit resolution', () => { it('stamps a request-supplied limit onto the inserted row', async () => { await provisionWithConcurrencyLimit(2); @@ -788,6 +736,7 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [10], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: 2, @@ -815,24 +764,12 @@ describe('DurableJobProvisioner', () => { expect(jobs.updateRunOptions).toHaveBeenCalledWith(manager, [], expect.anything()); }); - it('stores the largest limit the column holds', async () => { - await provisionWithConcurrencyLimit(2_147_483_647); + it('rejects an out-of-range limit with a user error', async () => { + await expect(provisionWithConcurrencyLimit(0)).rejects.toThrow(UserError); - expect(jobs.insertMany).toHaveBeenCalledWith(manager, [ - expect.objectContaining({ concurrencyLimit: 2_147_483_647 }), - ]); + expect(jobs.insertMany).not.toHaveBeenCalled(); + expect(dataSource.transaction).not.toHaveBeenCalled(); }); - - // 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', () => { @@ -859,6 +796,7 @@ describe('DurableJobProvisioner', () => { taskType: 'schedule-trigger', payload: {}, maxAttempts: 1, + timeoutSeconds: 45, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, }) as unknown as ScheduledJob; @@ -873,6 +811,7 @@ describe('DurableJobProvisioner', () => { scheduledFor: at(30), runAt: at(30), maxAttempts: 1, + timeoutSeconds: 45, // Its own instant plus the job's 60s grace. missedAfter: at(90), }, @@ -883,6 +822,7 @@ describe('DurableJobProvisioner', () => { scheduledFor: at(60), runAt: at(60), maxAttempts: 1, + timeoutSeconds: 45, missedAfter: at(120), }, ]; @@ -1036,6 +976,7 @@ describe('DurableJobProvisioner', () => { fireAt: null, nextRunAt: CLOCK, maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 90, concurrencyLimit: null, diff --git a/packages/cli/src/scheduling/durable-job-provisioner.ts b/packages/cli/src/scheduling/durable-job-provisioner.ts index 288409e1759..8dc270bf8e1 100644 --- a/packages/cli/src/scheduling/durable-job-provisioner.ts +++ b/packages/cli/src/scheduling/durable-job-provisioner.ts @@ -1,6 +1,6 @@ import { Logger } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; -import { MAX_INTEGER_32BITS_SIGNED, type ScheduledJobMisfirePolicy, Time } from '@n8n/constants'; +import type { ScheduledJobMisfirePolicy } from '@n8n/constants'; import type { EntityManager, NewScheduledJob, @@ -9,19 +9,26 @@ import type { } from '@n8n/db'; import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db'; import { Service } from '@n8n/di'; -import { createJobProvisioner, withOwnerKeys } from '@n8n/scheduler'; +import { + createJobProvisioner, + findOutdatedJobs, + InvalidRunOptionError, + resolveCoreRunOptions, + withOwnerKeys, +} from '@n8n/scheduler'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import type { DesiredJob, ExistingJob, JobProvisioner, ProvisionSummary, + ResolvedRunOptions, + RunOptions, RunInDeprovisionTransaction, RunInProvisionTransaction, } 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'; import { rowSchedule, scheduleColumns } from './schedule-columns'; @@ -30,19 +37,6 @@ import { createSchedulerTracer } from './scheduler-tracer'; import { SystemTaskScheduledJobOwner } from './system-tasks/system-task-scheduled-job-owner'; import { WorkflowScheduledJobOwner } from './workflow-scheduled-job-owner'; -/** - * Ceiling for a resolved misfire grace: the cap the config value carries, and well - * inside the column's `int` range. - */ -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. */ @@ -61,6 +55,11 @@ export interface ProvisionRequest { misfireGraceSeconds?: number; /** Retry ceiling stamped on each occurrence; omit to inherit the instance setting. */ maxAttempts?: number; + /** + * How long one run of an occurrence may take before the executor stops it. + * Omit to inherit the instance setting. + */ + timeoutSeconds?: number; /** * How many of the job's occurrences may run at the same time. Omit or pass * `null` for no limit. @@ -175,6 +174,7 @@ export class DurableJobProvisioner { * @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. + * @throws {UserError} when the timeout is not a whole number from 1 to the timer limit. * @returns what the call inserted, redefined, left unchanged and removed. */ async provision({ desired, ...scope }: ProvisionRequest): Promise { @@ -248,17 +248,9 @@ export class DurableJobProvisioner { owner, taskType, payload, - misfirePolicy, - misfireGraceSeconds: requestedMisfireGraceSeconds, - maxAttempts: requestedMaxAttempts, - concurrencyLimit: requestedConcurrencyLimit, + ...requested }: ProvisionScope): RunInProvisionTransaction { - const misfireGraceSeconds = this.resolveMisfireGraceSeconds( - requestedMisfireGraceSeconds, - owner, - ); - const maxAttempts = requestedMaxAttempts ?? this.globalConfig.scheduler.maxAttempts; - const concurrencyLimit = resolveConcurrencyLimit(requestedConcurrencyLimit); + const runOptions = this.resolveRunOptions(requested, owner); return async (work) => await this.dataSource.transaction(async (manager) => { // Provisioning is evidence the owner is back, so lift any quarantine now @@ -272,43 +264,25 @@ export class DurableJobProvisioner { }); } const existingRows = await this.jobs.findManyByOwner(manager, owner); - const outdatedRunOptionJobIds: number[] = []; - const outdatedGraceJobIds: number[] = []; - const outdatedPayloadJobIds: number[] = []; - for (const row of existingRows) { - const graceChanged = row.misfireGraceSeconds !== misfireGraceSeconds; - if (graceChanged) { - outdatedGraceJobIds.push(row.id); - } - if ( - graceChanged || - row.misfirePolicy !== misfirePolicy || - row.maxAttempts !== maxAttempts || - row.concurrencyLimit !== concurrencyLimit - ) { - outdatedRunOptionJobIds.push(row.id); - } - if (!isDeepStrictEqual(row.payload, payload)) { - outdatedPayloadJobIds.push(row.id); - } - } + const outdated = findOutdatedJobs(existingRows, runOptions, payload); // Before the provision, whose seed reads these rows back and must see the // current run options and payload. // Only `redefine` touches a job's run options, so an unchanged schedule // needs this to pick up a change to them on its own. - await this.jobs.updateRunOptions(manager, outdatedRunOptionJobIds, { - maxAttempts, - misfirePolicy, - misfireGraceSeconds, - concurrencyLimit, - }); + await this.jobs.updateRunOptions(manager, outdated.runOptions, runOptions); // Only `insert` writes the payload, so an existing row picks up a change to it here. - await this.jobs.updatePayload(manager, outdatedPayloadJobIds, payload); + await this.jobs.updatePayload(manager, outdated.payload, payload); // Queued tasks were stamped with the previous grace; recompute their deadline. await this.tasks.updateMissedAfterForJobs( manager, - outdatedGraceJobIds, - misfireGraceSeconds, + outdated.misfireGrace, + runOptions.misfireGraceSeconds, + ); + // A running occurrence keeps the timeout it was claimed with. + await this.tasks.updateTimeoutForPendingJobs( + manager, + outdated.timeout, + runOptions.timeoutSeconds, ); return await work({ findExisting: async () => @@ -329,10 +303,7 @@ export class DurableJobProvisioner { payload, ...scheduleColumns(job.schedule), nextRunAt: job.firstRunAt, - maxAttempts, - misfirePolicy, - misfireGraceSeconds, - concurrencyLimit, + ...runOptions, }), ); return await this.jobs.insertMany(manager, rows); @@ -341,10 +312,7 @@ export class DurableJobProvisioner { await this.jobs.updateDefinition(manager, jobId, { ...scheduleColumns(schedule), nextRunAt, - maxAttempts, - misfirePolicy, - misfireGraceSeconds, - concurrencyLimit, + ...runOptions, }); }, withdrawPendingTasks: async (jobIds) => @@ -374,45 +342,49 @@ export class DurableJobProvisioner { }); } - private resolveMisfireGraceSeconds(requested: unknown, owner: ScheduledJobOwner): number { - const { misfireGraceSeconds, executorIntervalSeconds, materializationWindowSeconds } = - this.globalConfig.scheduler; - - const numeric = Number(requested); - if (!Number.isFinite(numeric)) { - return misfireGraceSeconds; + /** + * Resolve a request against the instance settings, and warn when the requested + * misfire grace was moved to a bound. + * + * @throws {UserError} when the timeout or the concurrency limit is out of range. + */ + private resolveRunOptions( + requested: Omit, + owner: ScheduledJobOwner, + ): RunOptions { + const { scheduler } = this.globalConfig; + let resolved: ResolvedRunOptions; + try { + resolved = resolveCoreRunOptions(requested, { + maxAttempts: scheduler.maxAttempts, + timeoutSeconds: scheduler.taskTimeoutSeconds, + misfireGraceSeconds: scheduler.misfireGraceSeconds, + executorIntervalSeconds: scheduler.executorIntervalSeconds, + materializationWindowSeconds: scheduler.materializationWindowSeconds, + }); + } catch (error) { + if (error instanceof InvalidRunOptionError) { + throw new UserError(error.message, { + cause: error, + extra: { option: error.option, value: error.value, max: error.max }, + }); + } + throw error; } - - const truncated = Math.trunc(numeric); - if (truncated < 1) { - return misfireGraceSeconds; - } - - const floor = Math.min( - Math.max(executorIntervalSeconds + 1, materializationWindowSeconds), - MAX_MISFIRE_GRACE_SECONDS, - ); - - if (!Number.isFinite(floor)) { - return misfireGraceSeconds; - } - - const effective = Math.min(Math.max(truncated, floor), MAX_MISFIRE_GRACE_SECONDS); - - if (effective !== truncated || numeric > MAX_MISFIRE_GRACE_SECONDS) { + const { runOptions, misfireGraceAdjustment } = resolved; + if (misfireGraceAdjustment) { this.logger.warn( - effective > truncated + misfireGraceAdjustment.direction === 'raised' ? "Raised a node's misfire grace to the scheduler's minimum" : "Lowered a node's misfire grace to the scheduler's maximum", { ...owner, - requestedMisfireGraceSeconds: numeric, - misfireGraceSeconds: effective, + requestedMisfireGraceSeconds: misfireGraceAdjustment.requestedMisfireGraceSeconds, + misfireGraceSeconds: runOptions.misfireGraceSeconds, }, ); } - - return effective; + return runOptions; } private deprovisionTransaction(scope: DeprovisionScope): RunInDeprovisionTransaction { @@ -438,23 +410,3 @@ export class DurableJobProvisioner { } } } - -/** - * Normalize a requested concurrency ceiling to what the column stores: a whole - * number from 1 to {@link MAX_CONCURRENCY_LIMIT}, or `null` for no limit. - * - * @throws {UserError} when the limit is set but falls outside that range. A ceiling - * below one would hold every occurrence back until its misfire deadline passed, so - * the job would never run. - */ -function resolveConcurrencyLimit(requested: number | null | undefined): number | null { - if (requested === undefined || requested === null) { - return null; - } - if (!Number.isInteger(requested) || requested < 1 || requested > MAX_CONCURRENCY_LIMIT) { - throw new UserError('Scheduled job concurrency limit is outside the range the column holds', { - extra: { concurrencyLimit: requested, maxConcurrencyLimit: MAX_CONCURRENCY_LIMIT }, - }); - } - return requested; -} diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-job-registrar.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-job-registrar.test.ts index d2cd0c23c97..1819a3e4799 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-job-registrar.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-job-registrar.test.ts @@ -21,6 +21,7 @@ const NEXT_TEN = new Date('2026-01-05T10:00:00.000Z'); const WORKFLOW_ID = 'wf-1'; const NODE_ID = 'node-1'; const TIMEZONE = 'UTC'; +const POLL_TIMEOUT_SECONDS = 45; // Custom crons map straight through to `cron` schedules, so their first fire is exact. const DAILY_AT_NINE: TriggerTime = { mode: 'custom', cronExpression: '0 0 9 * * *' }; @@ -42,7 +43,10 @@ describe('PollTriggerJobRegistrar', () => { const makeRegistrar = () => new PollTriggerJobRegistrar( mockLogger(), - mock({ generic: { timezone: TIMEZONE } }), + mock({ + generic: { timezone: TIMEZONE }, + scheduler: { pollTimeoutSeconds: POLL_TIMEOUT_SECONDS }, + }), jobProvisioner, pollBackoffService, owner, @@ -96,6 +100,7 @@ describe('PollTriggerJobRegistrar', () => { }, ], misfirePolicy: ScheduledJobMisfirePolicy.Skip, + timeoutSeconds: POLL_TIMEOUT_SECONDS, }); }); @@ -246,6 +251,7 @@ describe('PollTriggerJobRegistrar', () => { payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID }, desired: [], misfirePolicy: ScheduledJobMisfirePolicy.Skip, + timeoutSeconds: POLL_TIMEOUT_SECONDS, }); }); diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts index 566b56542b0..8602d3b3e09 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts @@ -101,6 +101,7 @@ describe('PollTriggerTaskHandler', () => { scheduledFor, runAt: scheduledFor, status: 'running', + timeoutSeconds: 45, attempts: 0, maxAttempts: 1, leaseEpoch: 1, diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-job-registrar.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-job-registrar.ts index 4df15a9005c..6098307e06e 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-job-registrar.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-job-registrar.ts @@ -27,6 +27,8 @@ export class PollTriggerJobRegistrar extends PollJobManager { /** Instance-default timezone, used to resolve the `'DEFAULT'`/empty sentinel. */ private readonly defaultTimezone: string; + private readonly pollTimeoutSeconds: number; + constructor( private readonly logger: Logger, globalConfig: GlobalConfig, @@ -36,6 +38,7 @@ export class PollTriggerJobRegistrar extends PollJobManager { ) { super(); this.defaultTimezone = globalConfig.generic.timezone; + this.pollTimeoutSeconds = globalConfig.scheduler.pollTimeoutSeconds; this.logger = this.logger.scoped('scheduler'); } @@ -63,6 +66,7 @@ export class PollTriggerJobRegistrar extends PollJobManager { payload: { ...payload }, desired, misfirePolicy: ScheduledJobMisfirePolicy.Skip, + timeoutSeconds: this.pollTimeoutSeconds, }); this.logger.debug('Provisioned scheduler jobs for poll trigger node', { diff --git a/packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-task-handler.test.ts b/packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-task-handler.test.ts index 015e84d2229..d907f1324a4 100644 --- a/packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-task-handler.test.ts +++ b/packages/cli/src/scheduling/schedule-trigger-node/__tests__/schedule-trigger-task-handler.test.ts @@ -85,6 +85,7 @@ describe('ScheduleTriggerTaskHandler', () => { status: 'running', attempts: 0, maxAttempts: 1, + timeoutSeconds: 300, leaseEpoch: 1, ...overrides, }); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-job-registrar.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-job-registrar.test.ts index f8218c6b0fb..ee21f283a6d 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-job-registrar.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-job-registrar.test.ts @@ -122,6 +122,14 @@ describe('systemTaskProvisionRequest', () => { expect(request({ concurrencyLimit }).concurrencyLimit).toBe(concurrencyLimit); }); + it('leaves the timeout to the instance setting when the task declares none', () => { + expect(request()).toHaveProperty('timeoutSeconds', undefined); + }); + + it("honours a task's own timeout", () => { + expect(request({ timeoutSeconds: 600 }).timeoutSeconds).toBe(600); + }); + it("honours a task's own policy, grace and attempts", () => { expect( request({ diff --git a/packages/cli/src/scheduling/system-tasks/system-task-job-registrar.ts b/packages/cli/src/scheduling/system-tasks/system-task-job-registrar.ts index 8fdd158d629..d6902f427bf 100644 --- a/packages/cli/src/scheduling/system-tasks/system-task-job-registrar.ts +++ b/packages/cli/src/scheduling/system-tasks/system-task-job-registrar.ts @@ -37,7 +37,7 @@ export function systemTaskProvisionRequest( ? now : computeFirstRunAt(scheduleFromDefinition(schedule, defaultTimezone), now); const name = systemTaskType(task.name); - const { misfirePolicy, misfireGraceSeconds, maxAttempts, concurrencyLimit } = + const { misfirePolicy, misfireGraceSeconds, maxAttempts, concurrencyLimit, timeoutSeconds } = resolveSystemTaskRunOptions(task); return { @@ -49,6 +49,7 @@ export function systemTaskProvisionRequest( misfireGraceSeconds, maxAttempts, concurrencyLimit, + timeoutSeconds, }; } diff --git a/packages/cli/test/integration/scheduling/executor-repository.test.ts b/packages/cli/test/integration/scheduling/executor-repository.test.ts index f0fea151939..96e75b90db1 100644 --- a/packages/cli/test/integration/scheduling/executor-repository.test.ts +++ b/packages/cli/test/integration/scheduling/executor-repository.test.ts @@ -577,6 +577,7 @@ describe('ScheduledTaskRepository executor methods', () => { scheduledFor: earlier, runAt: earlier, maxAttempts: 1, + timeoutSeconds: 300, missedAfter: null, }, ]), diff --git a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts index 8f7a66ae2cc..d88c09089f9 100644 --- a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts +++ b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts @@ -103,6 +103,7 @@ describe('scheduled repositories', () => { fireAt: null, nextRunAt: secondsFromNow(-60), maxAttempts: 1, + timeoutSeconds: 300, ...overrides, }); @@ -144,6 +145,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }, ]); }); @@ -168,6 +170,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }, ]); await jobRepository.advanceMany(trx, [ @@ -199,6 +202,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }, ]), ).rejects.toThrow('insertIgnoringDuplicates must run within a transaction'); @@ -584,6 +588,7 @@ describe('scheduled repositories', () => { async (trx) => await jobRepository.updateRunOptions(trx, [updated.id], { maxAttempts: 5, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 120, concurrencyLimit: null, @@ -610,6 +615,7 @@ describe('scheduled repositories', () => { async (trx) => await jobRepository.updateRunOptions(trx, [job.id], { maxAttempts: job.maxAttempts, + timeoutSeconds: 300, misfirePolicy: job.misfirePolicy, misfireGraceSeconds: job.misfireGraceSeconds, concurrencyLimit: 3, @@ -628,6 +634,7 @@ describe('scheduled repositories', () => { async (trx) => await jobRepository.updateRunOptions(trx, [job.id], { maxAttempts: job.maxAttempts, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 60, concurrencyLimit: null, @@ -644,6 +651,7 @@ describe('scheduled repositories', () => { async (trx) => await jobRepository.updateRunOptions(trx, [], { maxAttempts: job.maxAttempts, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Skip, misfireGraceSeconds: 120, concurrencyLimit: null, @@ -768,6 +776,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }, ]), ); @@ -794,6 +803,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }, ]), ); @@ -817,6 +827,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }; const first = await dataSource.transaction( @@ -844,6 +855,7 @@ describe('scheduled repositories', () => { scheduledFor: when, runAt: when, maxAttempts: 1, + timeoutSeconds: 300, }; }); @@ -869,6 +881,7 @@ describe('scheduled repositories', () => { scheduledFor, runAt: scheduledFor, maxAttempts: 1, + timeoutSeconds: 300, }; const [first, second] = await Promise.all([ diff --git a/packages/cli/test/integration/scheduling/scheduler-benchmarks.test.ts b/packages/cli/test/integration/scheduling/scheduler-benchmarks.test.ts index 81770d2c0f8..848bc1ad824 100644 --- a/packages/cli/test/integration/scheduling/scheduler-benchmarks.test.ts +++ b/packages/cli/test/integration/scheduling/scheduler-benchmarks.test.ts @@ -256,6 +256,7 @@ describe.runIf(runBenchmarks)('durable scheduler benchmarks', () => { scheduledFor: when, runAt: when, maxAttempts: 1, + timeoutSeconds: 300, }; }); for (let i = 0; i < occurrences.length; i += SEED_TX_CHUNK) { 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 af07aa66528..056718ecba4 100644 --- a/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts +++ b/packages/cli/test/integration/scheduling/scheduler-query-benchmarks.test.ts @@ -512,6 +512,7 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => { scheduledFor: when, runAt: when, maxAttempts: 1, + timeoutSeconds: 300, }; }); const inserted = await dataSource.transaction( @@ -544,6 +545,7 @@ describe.runIf(runBenchmarks)('durable scheduler query benchmarks', () => { fireAt: null, nextRunAt: secondsFromNow(3600), maxAttempts: 1, + timeoutSeconds: 300, misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, misfireGraceSeconds: 60, concurrencyLimit: null, diff --git a/packages/cli/test/integration/scheduling/system-task-provisioning.test.ts b/packages/cli/test/integration/scheduling/system-task-provisioning.test.ts index 9e61b4638b6..68d76a6c077 100644 --- a/packages/cli/test/integration/scheduling/system-task-provisioning.test.ts +++ b/packages/cli/test/integration/scheduling/system-task-provisioning.test.ts @@ -91,6 +91,31 @@ describe('system task provisioning', () => { expect(seeded.every((occurrence) => occurrence.status === 'pending')).toBe(true); }); + it('applies a changed timeout to the job and its pending occurrences, not to a running one', async () => { + const schedule: SystemTaskSchedule = { kind: 'interval', intervalSeconds: 10 }; + await provision({ schedule }); + const inserted = await jobRepo.findOneByOrFail({ name: JOB_NAME }); + expect(inserted.timeoutSeconds).toBe(Container.get(GlobalConfig).scheduler.taskTimeoutSeconds); + const [running, ...pending] = await taskRepo.findBy({ jobId: inserted.id }); + expect(pending).not.toHaveLength(0); + await taskRepo.update( + { id: running.id }, + { status: 'running', claimedBy: 'main-a', leaseExpiresAt: new Date(Date.now() + 60_000) }, + ); + + const summary = await provision({ schedule, timeoutSeconds: 600 }); + + expect(summary.unchanged).toEqual([{ id: inserted.id, name: JOB_NAME }]); + const row = await jobRepo.findOneByOrFail({ name: JOB_NAME }); + expect(row.timeoutSeconds).toBe(600); + const tasks = await taskRepo.findBy({ jobId: inserted.id }); + const timeoutById = new Map(tasks.map((t) => [t.id, t.timeoutSeconds])); + expect(timeoutById.get(running.id)).toBe(inserted.timeoutSeconds); + for (const occurrence of pending) { + expect(timeoutById.get(occurrence.id)).toBe(600); + } + }); + it('seeds a claimable occurrence at once for an interval task', async () => { const before = new Date();