feat(core): Store and provision the timeout of scheduled jobs (no-changelog) (#40513)

This commit is contained in:
Lorent Lempereur 2026-10-12 10:46:10 +00:00 committed by GitHub
parent 2a1fde3634
commit 53a6019ad5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
43 changed files with 823 additions and 275 deletions

View File

@ -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.

View File

@ -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.

View File

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

View File

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

View File

@ -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<ScheduledTask> {
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<void> {
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

View File

@ -46,6 +46,7 @@ describe('executor claims far enough ahead to fire on time', () => {
status: 'pending',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 0,
});

View File

@ -250,6 +250,7 @@ const claimedTask = (overrides: Partial<ClaimedTask> = {}): 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,

View File

@ -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) {

View File

@ -20,6 +20,7 @@ const claimedTask = (id: string): ClaimedTask => ({
status: 'running',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 1,
});

View File

@ -23,6 +23,7 @@ const claimedTask = (overrides: Partial<ClaimedTask> = {}): ClaimedTask => ({
status: 'running',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 1,
...overrides,
});

View File

@ -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';

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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

View File

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

View File

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

View File

@ -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<RequestedRunOptions> = {},
defaults: Partial<RunOptionDefaults> = {},
) =>
resolveCoreRunOptions(
{ misfirePolicy: ScheduledJobMisfirePolicy.Coalesce, ...requested },
{ ...DEFAULTS, ...defaults },
);
const resolveGrace = (misfireGraceSeconds: unknown, defaults: Partial<RunOptionDefaults> = {}) =>
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<RunOptions & { payload: Record<string, unknown> }> = {}) => ({
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] });
});
});

View File

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

View File

@ -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<Pick<ScheduledJob, 'id' | 'payload'> & RunOptions>,
runOptions: RunOptions,
payload: Record<string, unknown>,
): 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;
}

View File

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

View File

@ -24,6 +24,7 @@ const expiredTask = (overrides: Partial<ExpiredLeaseRow> = {}): ExpiredLeaseRow
status: 'running',
attempts: 0,
maxAttempts: 3,
timeoutSeconds: 300,
leaseEpoch: 1,
// Pre-dispatch by default; post-dispatch tests set a concrete `dispatchedAt`.
dispatchedAt: null,

View File

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

View File

@ -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.

View File

@ -39,6 +39,7 @@ export interface ScheduledTask {
status: ScheduledTaskStatus;
attempts: number;
maxAttempts: number;
timeoutSeconds: number;
}
/**

View File

@ -20,6 +20,7 @@ const claimedTask: ClaimedTask = {
status: 'running',
attempts: 1,
maxAttempts: 3,
timeoutSeconds: 300,
leaseEpoch: 2,
};

View File

@ -19,6 +19,7 @@ const claimedTask = (overrides: Partial<ClaimedTask> = {}): ClaimedTask => ({
status: 'running',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 1,
...overrides,
});

View File

@ -34,6 +34,7 @@ describe('AgentTaskTaskHandler', () => {
status: 'running',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 1,
...overrides,
});

View File

@ -53,6 +53,7 @@ const jobRow = ({ payload = {}, ...over }: Partial<ScheduledJob> = {}): 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<ProvisionSummary> =>
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,

View File

@ -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<ProvisionSummary> {
@ -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<ProvisionScope, 'owner' | 'taskType' | 'payload'>,
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;
}

View File

@ -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<GlobalConfig>({ generic: { timezone: TIMEZONE } }),
mock<GlobalConfig>({
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,
});
});

View File

@ -101,6 +101,7 @@ describe('PollTriggerTaskHandler', () => {
scheduledFor,
runAt: scheduledFor,
status: 'running',
timeoutSeconds: 45,
attempts: 0,
maxAttempts: 1,
leaseEpoch: 1,

View File

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

View File

@ -85,6 +85,7 @@ describe('ScheduleTriggerTaskHandler', () => {
status: 'running',
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 300,
leaseEpoch: 1,
...overrides,
});

View File

@ -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({

View File

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

View File

@ -577,6 +577,7 @@ describe('ScheduledTaskRepository executor methods', () => {
scheduledFor: earlier,
runAt: earlier,
maxAttempts: 1,
timeoutSeconds: 300,
missedAfter: null,
},
]),

View File

@ -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([

View File

@ -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) {

View File

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

View File

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