diff --git a/packages/@n8n/scheduler/README.md b/packages/@n8n/scheduler/README.md new file mode 100644 index 00000000000..da3d9bb6a9a --- /dev/null +++ b/packages/@n8n/scheduler/README.md @@ -0,0 +1,28 @@ +# @n8n/scheduler + +A durable, distributed scheduling service for n8n. It schedules recurring and +one-off work and dispatches each due occurrence to a handler, coordinating through +the database so every main instance participates and no occurrence is lost across a +restart or failover. Scheduling (deciding that something is due) is decoupled from +execution (running it), and dispatch is effectively-once. + +It is a general scheduling primitive, not tied to one feature: each occurrence is +routed to a handler by a `task_type` key. The Schedule Trigger node is its first +consumer; poll triggers, system tasks and waiting executions are intended to follow, +replacing today's in-memory, leader-only scheduling. All of it sits behind +`N8N_SCHEDULER_ENABLED` (default off) while the legacy engine stays in place for +rollback. + +## Scope + +- **Coordination** (the core): assign each due occurrence to a single main via + claim, lease and fencing, recover work whose owner has died, and dispatch it + across the cluster. +- **Recurrence**: turn a cron, interval or one-off schedule definition into its next + occurrence, with correct timezone and DST handling. + +## Status + +Early foundation. Today the package ships the domain types, the schedule math and a +thin storage boundary; the coordination engine (sweep, executor, reaper) lands in +later milestones. diff --git a/packages/@n8n/scheduler/eslint.config.mjs b/packages/@n8n/scheduler/eslint.config.mjs new file mode 100644 index 00000000000..541ae772bb0 --- /dev/null +++ b/packages/@n8n/scheduler/eslint.config.mjs @@ -0,0 +1,8 @@ +import { defineConfig } from 'eslint/config'; +import { baseConfig } from '@n8n/eslint-config/base'; + +export default defineConfig(baseConfig, { + rules: { + 'unicorn/filename-case': ['error', { case: 'kebabCase' }], + }, +}); diff --git a/packages/@n8n/scheduler/package.json b/packages/@n8n/scheduler/package.json new file mode 100644 index 00000000000..b39109bd61e --- /dev/null +++ b/packages/@n8n/scheduler/package.json @@ -0,0 +1,39 @@ +{ + "name": "@n8n/scheduler", + "version": "0.1.0", + "scripts": { + "clean": "rimraf dist .turbo", + "dev": "pnpm watch", + "typecheck": "tsc --noEmit", + "build": "tsc -p tsconfig.build.json", + "build:unchecked": "tsc -p tsconfig.build.json --noCheck", + "format": "biome format --write .", + "format:check": "biome ci .", + "lint": "eslint . --quiet", + "lint:fix": "eslint . --fix", + "watch": "tsc -p tsconfig.build.json --watch", + "test": "vitest run --passWithNoTests", + "test:unit": "vitest run --passWithNoTests", + "test:dev": "vitest --silent=false" + }, + "main": "dist/index.js", + "module": "src/index.ts", + "types": "dist/index.d.ts", + "files": [ + "dist/**/*" + ], + "dependencies": { + "cron-parser": "catalog:", + "luxon": "catalog:", + "n8n-workflow": "workspace:*" + }, + "devDependencies": { + "@n8n/typescript-config": "workspace:*", + "@n8n/vitest-config": "workspace:*", + "@types/luxon": "catalog:", + "@vitest/coverage-v8": "catalog:", + "vitest": "catalog:", + "vitest-mock-extended": "catalog:" + }, + "license": "LicenseRef-n8n-sustainable-use" +} diff --git a/packages/@n8n/scheduler/src/errors.ts b/packages/@n8n/scheduler/src/errors.ts new file mode 100644 index 00000000000..a78c77dbfe7 --- /dev/null +++ b/packages/@n8n/scheduler/src/errors.ts @@ -0,0 +1,14 @@ +/** + * Raised when a schedule fails validation (bad cron expression, non-positive + * interval, invalid one-off instant, unresolved timezone, unknown kind). + * + * A domain-specific error rather than n8n's `UserError`: it keeps schedule + * failures identifiable on their own and the package decoupled from the n8n + * error hierarchy. + */ +export class InvalidScheduleError extends Error { + constructor(message: string) { + super(message); + this.name = 'InvalidScheduleError'; + } +} diff --git a/packages/@n8n/scheduler/src/index.ts b/packages/@n8n/scheduler/src/index.ts new file mode 100644 index 00000000000..a02b2c58bbf --- /dev/null +++ b/packages/@n8n/scheduler/src/index.ts @@ -0,0 +1,33 @@ +/** + * `@n8n/scheduler`: durable, multi-main work scheduler. + * + * The core concern is *coordination*: each unit of scheduled work (a + * `ScheduledTask`) is claimed and run exactly once on one main, with leases and + * fencing for crash recovery. That layer is time-agnostic; it only asks whether a + * task is due now. The claim / lease / fencing / reaper code lands in later + * tickets. + * + * *Recurrence* is one source of work, not the core: a `ScheduledJob` with a cron + * / interval / one-off `Schedule` is materialised into tasks. All time and DST + * math is confined to this boundary (see `recurrence/`). + */ + +export { ScheduleKindList, TaskStatusList } from './types'; + +export type { + ScheduleKind, + CronSchedule, + IntervalSchedule, + OneOffSchedule, + Schedule, + TaskStatus, + ScheduledJob, + ScheduledTask, +} from './types'; + +export { InvalidScheduleError } from './errors'; + +export { computeNextRunAt } from './recurrence/next-run'; +export { validateSchedule } from './recurrence/validate'; + +export type { SchedulerStore } from './storage/storage'; diff --git a/packages/@n8n/scheduler/src/recurrence/__tests__/next-run.test.ts b/packages/@n8n/scheduler/src/recurrence/__tests__/next-run.test.ts new file mode 100644 index 00000000000..354632af5d2 --- /dev/null +++ b/packages/@n8n/scheduler/src/recurrence/__tests__/next-run.test.ts @@ -0,0 +1,183 @@ +import { DateTime } from 'luxon'; + +import { InvalidScheduleError } from '../../errors'; +import type { CronSchedule, Schedule } from '../../types'; +import { computeNextRunAt } from '../next-run'; + +/** computeNextRunAt asserting a non-null result, for the unbounded (cron/interval) cases. */ +const nextOf = (schedule: Schedule, after: Date): Date => { + const next = computeNextRunAt(schedule, after); + if (next === null) throw new Error('expected a next run, got null'); + return next; +}; + +/** Local wall-clock ISO (with offset) of an absolute instant, in a given zone. */ +const localOf = (d: Date, zone: string) => + DateTime.fromJSDate(d).setZone(zone).toISO({ suppressMilliseconds: true }); + +describe('computeNextRunAt', () => { + describe('cron', () => { + const utcDaily: CronSchedule = { kind: 'cron', cronExpression: '0 0 0 * * *', timezone: 'UTC' }; + + it('computes the next daily fire strictly after the base', () => { + expect(nextOf(utcDaily, new Date('2026-01-10T12:00:00Z')).toISOString()).toBe( + '2026-01-11T00:00:00.000Z', + ); + }); + + it('skips the base instant when it lands exactly on a fire (strictly-after)', () => { + expect(nextOf(utcDaily, new Date('2026-01-10T00:00:00Z')).toISOString()).toBe( + '2026-01-11T00:00:00.000Z', + ); + }); + + it('honours the seconds field (6-field)', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '30 * * * * *', + timezone: 'UTC', + }; + expect(nextOf(schedule, new Date('2026-01-10T00:00:00Z')).toISOString()).toBe( + '2026-01-10T00:00:30.000Z', + ); + }); + + it('throws when the timezone is unresolved (null)', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '0 0 0 * * *', + timezone: null, + }; + expect(() => computeNextRunAt(schedule, new Date('2026-01-10T12:00:00Z'))).toThrow( + InvalidScheduleError, + ); + }); + + it('throws InvalidScheduleError on an out-of-range expression', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '99 0 0 * * *', + timezone: 'UTC', + }; + expect(() => computeNextRunAt(schedule, new Date('2026-01-10T00:00:00Z'))).toThrow( + InvalidScheduleError, + ); + }); + }); + + describe('interval', () => { + it('advances by intervalSeconds from after', () => { + expect( + nextOf( + { kind: 'interval', intervalSeconds: 3600 }, + new Date('2026-01-01T00:00:00Z'), + ).toISOString(), + ).toBe('2026-01-01T01:00:00.000Z'); + }); + + it('advances from an arbitrary prior occurrence (deterministic, strictly after)', () => { + expect( + nextOf( + { kind: 'interval', intervalSeconds: 3600 }, + new Date('2026-01-01T02:30:00Z'), + ).toISOString(), + ).toBe('2026-01-01T03:30:00.000Z'); + }); + }); + + describe('one_off', () => { + const fireAt = new Date('2026-01-10T00:00:00.000Z'); + + it('returns fireAt when after is before it', () => { + expect( + computeNextRunAt({ kind: 'one_off', fireAt }, new Date('2026-01-09T23:59:59.999Z')), + ).toEqual(fireAt); + }); + + it('returns null when after is at or past fireAt (strictly-after, exhausted)', () => { + expect(computeNextRunAt({ kind: 'one_off', fireAt }, new Date(fireAt))).toBeNull(); + expect( + computeNextRunAt({ kind: 'one_off', fireAt }, new Date('2026-02-01T00:00:00Z')), + ).toBeNull(); + }); + }); + + // computeNextRunAt validates first, so corrupt input throws rather than + // returning a wrong, past, or Invalid instant. + describe('rejects malformed input', () => { + const after = new Date('2026-01-01T00:00:00Z'); + + it('throws on a zero interval (would not advance)', () => { + expect(() => computeNextRunAt({ kind: 'interval', intervalSeconds: 0 }, after)).toThrow( + InvalidScheduleError, + ); + }); + + it('throws on a negative interval (would go backwards)', () => { + expect(() => computeNextRunAt({ kind: 'interval', intervalSeconds: -60 }, after)).toThrow( + InvalidScheduleError, + ); + }); + + it('throws on a non-finite interval', () => { + expect(() => computeNextRunAt({ kind: 'interval', intervalSeconds: NaN }, after)).toThrow( + InvalidScheduleError, + ); + }); + + it('throws on an invalid one-off instant', () => { + expect(() => computeNextRunAt({ kind: 'one_off', fireAt: new Date('nope') }, after)).toThrow( + InvalidScheduleError, + ); + }); + }); + + describe('DST', () => { + // America/New_York springs forward 2026-03-08: 02:00 -> 03:00 (02:xx local does not exist). + it('cron spring-forward: shifts a nonexistent 02:30 local fire to 03:30 local', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '0 30 2 * * *', + timezone: 'America/New_York', + }; + const next = nextOf(schedule, new Date('2026-03-08T00:00:00-05:00')); + expect(next.toISOString()).toBe('2026-03-08T07:30:00.000Z'); + expect(localOf(next, 'America/New_York')).toBe('2026-03-08T03:30:00-04:00'); + }); + + // America/New_York falls back 2026-11-01: 02:00 -> 01:00 (01:xx local happens twice). + it('cron fall-back: fires the daily 01:30 once, then the next day (no double-fire)', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '0 30 1 * * *', + timezone: 'America/New_York', + }; + const first = nextOf(schedule, new Date('2026-11-01T00:00:00-04:00')); + expect(localOf(first, 'America/New_York')).toBe('2026-11-01T01:30:00-04:00'); + + const second = nextOf(schedule, first); + expect(localOf(second, 'America/New_York')).toBe('2026-11-02T01:30:00-05:00'); + }); + + it('interval across spring-forward stays 3600s apart (real elapsed, skips the wall hour)', () => { + // Prior occurrence 01:30 EST (06:30Z). +1h real -> 07:30Z = 03:30 EDT. + const next = nextOf( + { kind: 'interval', intervalSeconds: 3600 }, + new Date('2026-03-08T06:30:00Z'), + ); + expect(next.toISOString()).toBe('2026-03-08T07:30:00.000Z'); + expect(localOf(next, 'America/New_York')).toBe('2026-03-08T03:30:00-04:00'); + }); + + it('UTC control: a daily UTC cron is unaffected by any DST transition', () => { + const schedule: CronSchedule = { + kind: 'cron', + cronExpression: '0 0 12 * * *', + timezone: 'UTC', + }; + expect(nextOf(schedule, new Date('2026-03-08T00:00:00Z')).toISOString()).toBe( + '2026-03-08T12:00:00.000Z', + ); + }); + }); +}); diff --git a/packages/@n8n/scheduler/src/recurrence/__tests__/validate.test.ts b/packages/@n8n/scheduler/src/recurrence/__tests__/validate.test.ts new file mode 100644 index 00000000000..e2090970cc5 --- /dev/null +++ b/packages/@n8n/scheduler/src/recurrence/__tests__/validate.test.ts @@ -0,0 +1,94 @@ +import type { CronExpression } from 'n8n-workflow'; + +import { InvalidScheduleError } from '../../errors'; +import { validateSchedule } from '../validate'; + +describe('validateSchedule', () => { + describe('cron', () => { + it('accepts a valid 6-field expression in a real timezone', () => { + expect(() => + validateSchedule({ + kind: 'cron', + cronExpression: '0 0 12 * * *', + timezone: 'Europe/London', + }), + ).not.toThrow(); + }); + + it('accepts a null timezone (instance default)', () => { + expect(() => + validateSchedule({ kind: 'cron', cronExpression: '0 0 12 * * *', timezone: null }), + ).not.toThrow(); + }); + + it('rejects a 5-field expression (seconds field required)', () => { + expect(() => + validateSchedule({ + kind: 'cron', + cronExpression: '0 12 * * *' as unknown as CronExpression, + timezone: 'UTC', + }), + ).toThrow(InvalidScheduleError); + }); + + it('rejects an unknown timezone', () => { + expect(() => + validateSchedule({ kind: 'cron', cronExpression: '0 0 12 * * *', timezone: 'Mars/Phobos' }), + ).toThrow(/timezone/); + }); + + it('rejects an out-of-range expression', () => { + expect(() => + validateSchedule({ kind: 'cron', cronExpression: '99 0 0 * * *', timezone: 'UTC' }), + ).toThrow(InvalidScheduleError); + }); + + it('rejects a non-string cron expression (raw input) with InvalidScheduleError', () => { + expect(() => + validateSchedule({ + kind: 'cron', + cronExpression: null as unknown as CronExpression, + timezone: 'UTC', + }), + ).toThrow(InvalidScheduleError); + }); + }); + + describe('interval', () => { + it('accepts a positive intervalSeconds', () => { + expect(() => validateSchedule({ kind: 'interval', intervalSeconds: 60 })).not.toThrow(); + }); + + it('rejects a non-positive intervalSeconds', () => { + expect(() => validateSchedule({ kind: 'interval', intervalSeconds: 0 })).toThrow( + /intervalSeconds/, + ); + }); + + it('rejects a non-integer intervalSeconds', () => { + expect(() => validateSchedule({ kind: 'interval', intervalSeconds: 1.5 })).toThrow( + InvalidScheduleError, + ); + }); + }); + + describe('one_off', () => { + it('accepts a valid fireAt', () => { + expect(() => + validateSchedule({ kind: 'one_off', fireAt: new Date('2026-01-01T00:00:00Z') }), + ).not.toThrow(); + }); + + it('rejects an invalid fireAt', () => { + expect(() => validateSchedule({ kind: 'one_off', fireAt: new Date('nope') })).toThrow( + InvalidScheduleError, + ); + }); + + it('rejects a non-Date fireAt (raw input) with InvalidScheduleError', () => { + expect(() => + validateSchedule({ kind: 'one_off', fireAt: '2026-01-01T00:00:00Z' as unknown as Date }), + ).toThrow(InvalidScheduleError); + }); + }); +}); diff --git a/packages/@n8n/scheduler/src/recurrence/next-run.ts b/packages/@n8n/scheduler/src/recurrence/next-run.ts new file mode 100644 index 00000000000..572ec22459e --- /dev/null +++ b/packages/@n8n/scheduler/src/recurrence/next-run.ts @@ -0,0 +1,79 @@ +import { CronExpressionParser } from 'cron-parser'; + +import { InvalidScheduleError } from '../errors'; +import type { CronSchedule, IntervalSchedule, OneOffSchedule, Schedule } from '../types'; +import { validateSchedule } from './validate'; + +const MS_PER_SECOND = 1000; + +/** + * Cron: next fire strictly after `after`, in the schedule's IANA timezone. + * `cron-parser` advances from `currentDate` with strictly-after semantics and + * resolves DST via luxon. The timezone must already be resolved to a concrete + * zone (a `null` instance default is rejected upstream). Wall-clock: a + * nonexistent local time (spring-forward) shifts forward; a repeated local time + * (fall-back) fires once. + */ +function cronNextRun(schedule: CronSchedule, after: Date, timezone: string): Date { + try { + const it = CronExpressionParser.parse(schedule.cronExpression, { + currentDate: after, + tz: timezone, + }); + return it.next().toDate(); + } catch (error) { + throw new InvalidScheduleError( + `Failed to evaluate cron expression ${JSON.stringify(schedule.cronExpression)} in timezone ${JSON.stringify(timezone)}: ${(error as Error).message}`, + ); + } +} + +/** + * Interval: advances by `intervalSeconds` of real elapsed time (UTC) from + * `after` (the prior occurrence), so the cadence is deterministic and DST never + * shifts a fire. Always strictly after `after` (intervalSeconds is positive). + */ +function intervalNextRun(schedule: IntervalSchedule, after: Date): Date { + return new Date(after.getTime() + schedule.intervalSeconds * MS_PER_SECOND); +} + +/** One-off: `fireAt` when it is strictly after `after`, otherwise `null` (exhausted). */ +function oneOffNextRun(schedule: OneOffSchedule, after: Date): Date | null { + return after.getTime() < schedule.fireAt.getTime() ? schedule.fireAt : null; +} + +/** + * Compute the next occurrence strictly after `after` (a job's current + * `nextRunAt` / last scheduled instant), as a UTC instant. This is what the + * sweep advances `next_run_at` with. + * + * The schedule is validated first, so malformed input (non-positive interval, + * invalid `fireAt`, bad cron expression, unresolved `null` cron timezone) throws + * {@link InvalidScheduleError} rather than returning a wrong or `Invalid` instant. + * + * Returns `null` only when the schedule is exhausted (a one-off already at or + * past `after`); cron and interval schedules are unbounded. + */ +export function computeNextRunAt(schedule: Schedule, after: Date): Date | null { + validateSchedule(schedule); + + switch (schedule.kind) { + case 'cron': + if (schedule.timezone === null) { + throw new InvalidScheduleError( + 'Cron timezone must be resolved to a concrete zone before computing the next run, got null', + ); + } + return cronNextRun(schedule, after, schedule.timezone); + case 'interval': + return intervalNextRun(schedule, after); + case 'one_off': + return oneOffNextRun(schedule, after); + default: { + const exhaustive: never = schedule; + throw new InvalidScheduleError( + `Unknown schedule kind: ${JSON.stringify((exhaustive as Schedule).kind)}`, + ); + } + } +} diff --git a/packages/@n8n/scheduler/src/recurrence/validate.ts b/packages/@n8n/scheduler/src/recurrence/validate.ts new file mode 100644 index 00000000000..1487ce257a1 --- /dev/null +++ b/packages/@n8n/scheduler/src/recurrence/validate.ts @@ -0,0 +1,78 @@ +import { CronExpressionParser } from 'cron-parser'; +import { IANAZone } from 'luxon'; + +import { InvalidScheduleError } from '../errors'; +import type { CronSchedule, IntervalSchedule, OneOffSchedule, Schedule } from '../types'; + +/** Cron expressions are required to be 6-field (with a leading seconds field). */ +const CRON_FIELD_COUNT = 6; + +function validateCron(schedule: CronSchedule): void { + // This is the boundary for raw, possibly-untyped input (DB rows), so guard the + // runtime type before using it instead of throwing a raw TypeError. + const expression: unknown = schedule.cronExpression; + if (typeof expression !== 'string') { + throw new InvalidScheduleError( + `cron.cronExpression must be a string, got ${JSON.stringify(expression)}`, + ); + } + + const fieldCount = expression.trim().split(/\s+/).length; + if (fieldCount !== CRON_FIELD_COUNT) { + throw new InvalidScheduleError( + `Cron expression must have ${CRON_FIELD_COUNT} fields (seconds included), got ${fieldCount}: ${JSON.stringify(expression)}`, + ); + } + + // A null timezone is the instance default, resolved by the caller. + if (schedule.timezone !== null && !IANAZone.isValidZone(schedule.timezone)) { + throw new InvalidScheduleError(`Unknown IANA timezone: ${JSON.stringify(schedule.timezone)}`); + } + + try { + CronExpressionParser.parse(expression, { + tz: schedule.timezone ?? 'UTC', + }); + } catch (error) { + throw new InvalidScheduleError( + `Invalid cron expression ${JSON.stringify(expression)}: ${(error as Error).message}`, + ); + } +} + +function validateInterval(schedule: IntervalSchedule): void { + if (!Number.isInteger(schedule.intervalSeconds) || schedule.intervalSeconds <= 0) { + throw new InvalidScheduleError( + `interval.intervalSeconds must be a positive integer, got ${JSON.stringify(schedule.intervalSeconds)}`, + ); + } +} + +function validateOneOff(schedule: OneOffSchedule): void { + const fireAt: unknown = schedule.fireAt; + if (!(fireAt instanceof Date) || Number.isNaN(fireAt.getTime())) { + throw new InvalidScheduleError('one_off.fireAt must be a valid Date'); + } +} + +/** + * Validate a schedule definition, throwing {@link InvalidScheduleError} on the + * first problem. Safe to call before persisting a schedule or computing its next + * run. + */ +export function validateSchedule(schedule: Schedule): void { + switch (schedule.kind) { + case 'cron': + return validateCron(schedule); + case 'interval': + return validateInterval(schedule); + case 'one_off': + return validateOneOff(schedule); + default: { + const exhaustive: never = schedule; + throw new InvalidScheduleError( + `Unknown schedule kind: ${JSON.stringify((exhaustive as Schedule).kind)}`, + ); + } + } +} diff --git a/packages/@n8n/scheduler/src/storage/storage.ts b/packages/@n8n/scheduler/src/storage/storage.ts new file mode 100644 index 00000000000..87d4c5e8e3a --- /dev/null +++ b/packages/@n8n/scheduler/src/storage/storage.ts @@ -0,0 +1,34 @@ +import type { ScheduledJob, ScheduledTask } from '../types'; + +/** + * Storage port for the scheduler: the seam between the engine and persistence. + * + * Deliberately thin and provisional: it holds only what the schedule math and an + * early sweep imply. The real contract (claiming, leasing, fencing, retention) + * emerges with the storage adapter, which owns it. Do not grow this interface + * here. + */ +export interface SchedulerStore { + /** + * Fetch enabled jobs that are due to fire, for the sweep to materialise. + * + * @param now - Reference instant; a job is due when its `nextRunAt` is at or before this. + * @param limit - Maximum number of jobs to return, so one sweep claims a bounded batch. + * @returns The due jobs (at most `limit`). + */ + getDueJobs(now: Date, limit: number): Promise; + + /** + * Persist a job's advanced scheduling state after the sweep has fired it. + * + * @param job - The job carrying the updated `nextRunAt` and `lastFiredAt`. + */ + saveJob(job: ScheduledJob): Promise; + + /** + * Enqueue a materialised occurrence for the executor to claim and run. + * + * @param task - The occurrence to insert; identity is unique on `(jobId, scheduledFor)`. + */ + createTask(task: ScheduledTask): Promise; +} diff --git a/packages/@n8n/scheduler/src/types/index.ts b/packages/@n8n/scheduler/src/types/index.ts new file mode 100644 index 00000000000..e794ecb589f --- /dev/null +++ b/packages/@n8n/scheduler/src/types/index.ts @@ -0,0 +1,2 @@ +export * from './schedule'; +export * from './task'; diff --git a/packages/@n8n/scheduler/src/types/schedule.ts b/packages/@n8n/scheduler/src/types/schedule.ts new file mode 100644 index 00000000000..6d1ca81ca40 --- /dev/null +++ b/packages/@n8n/scheduler/src/types/schedule.ts @@ -0,0 +1,61 @@ +import type { CronExpression } from 'n8n-workflow'; + +/** + * The recurrence source: a `ScheduledJob` defines recurring work through a cron / + * interval / one-off `Schedule`, which the materialiser turns into tasks. Time + * and DST math over these types lives in `recurrence/`; coordination (the core) + * never looks at them. + * + * Field names mirror the `scheduled_job` columns so the storage adapter maps + * trivially. Instants are `Date` (absolute UTC). `CronExpression` is imported + * from `n8n-workflow` (its canonical home) and not re-exported here. + */ + +/** + * The recurrence kinds as a runtime list (not a bare union) so the schema column + * and validation share one source of truth. Same idiom as `ExecutionStatusList`. + */ +export const ScheduleKindList = ['cron', 'interval', 'one_off'] as const; +export type ScheduleKind = (typeof ScheduleKindList)[number]; + +/** + * A 6-field cron expression (seconds included) evaluated in an IANA timezone. + * Wall-clock: fires at the given local time, so DST shifts the absolute instant. + * `timezone === null` means the instance default, resolved by the caller before + * the math runs. + */ +export interface CronSchedule { + kind: 'cron'; + cronExpression: CronExpression; + timezone: string | null; +} + +/** + * A fixed-period schedule firing every `intervalSeconds`, measured as absolute + * elapsed time (UTC) from the prior occurrence, so DST never shifts a fire. No + * timezone field by design. + */ +export interface IntervalSchedule { + kind: 'interval'; + intervalSeconds: number; +} + +/** Fires exactly once at `fireAt`, then never again. A fixed instant: no tz/DST concern. */ +export interface OneOffSchedule { + kind: 'one_off'; + fireAt: Date; +} + +export type Schedule = CronSchedule | IntervalSchedule | OneOffSchedule; + +/** + * A schedule definition (`scheduled_job`). Recurrence lives in `schedule`; + * `nextRunAt` is the next instant the sweep materialises from. + */ +export interface ScheduledJob { + id: string; + schedule: Schedule; + enabled: boolean; + nextRunAt: Date | null; + lastFiredAt: Date | null; +} diff --git a/packages/@n8n/scheduler/src/types/task.ts b/packages/@n8n/scheduler/src/types/task.ts new file mode 100644 index 00000000000..22cf94072e4 --- /dev/null +++ b/packages/@n8n/scheduler/src/types/task.ts @@ -0,0 +1,42 @@ +/** + * The allocated unit of work. Coordination (the scheduler's core) claims, leases, + * and runs each `ScheduledTask` exactly once on one main; recurrence is just one + * way these come to exist. + * + * Field names mirror the `scheduled_task` columns so the storage adapter maps + * trivially. Instants are `Date` (absolute UTC). + */ + +/** + * Occurrence lifecycle (`scheduled_task.status`) as a runtime list, so the schema + * column and validation share one source of truth. + */ +export const TaskStatusList = [ + 'pending', + 'running', + 'succeeded', + 'failed', + 'missed', + 'cancelled', +] as const; +export type TaskStatus = (typeof TaskStatusList)[number]; + +/** + * A materialised occurrence of a job (`scheduled_task`): one queued run. + * + * `scheduledFor` is the canonical UTC occurrence instant and the identity (unique + * on `(jobId, scheduledFor)`); `runAt` is the visibility time (equal to + * `scheduledFor` initially, pushed forward by retry backoff). Coordination fields + * (owner, lease, fencing) are added with the claim and reaper work. + */ +export interface ScheduledTask { + id: string; + jobId: string; + taskType: string; + payload: unknown; + scheduledFor: Date; + runAt: Date; + status: TaskStatus; + attempts: number; + maxAttempts: number; +} diff --git a/packages/@n8n/scheduler/tsconfig.build.json b/packages/@n8n/scheduler/tsconfig.build.json new file mode 100644 index 00000000000..e08cb937d03 --- /dev/null +++ b/packages/@n8n/scheduler/tsconfig.build.json @@ -0,0 +1,10 @@ +{ + "extends": ["./tsconfig.json", "@n8n/typescript-config/tsconfig.build.json"], + "compilerOptions": { + "composite": true, + "rootDir": "src", + "outDir": "dist" + }, + "include": ["src/**/*.ts"], + "exclude": ["src/**/__tests__/**"] +} diff --git a/packages/@n8n/scheduler/tsconfig.json b/packages/@n8n/scheduler/tsconfig.json new file mode 100644 index 00000000000..6c32c1071e4 --- /dev/null +++ b/packages/@n8n/scheduler/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "@n8n/typescript-config/tsconfig.common.json", + "compilerOptions": { + "types": ["node", "vitest/globals"], + "paths": { + "@/*": ["./src/*"] + } + }, + "include": ["src/**/*.ts"] +} diff --git a/packages/@n8n/scheduler/vite.config.ts b/packages/@n8n/scheduler/vite.config.ts new file mode 100644 index 00000000000..bcfcc92b9f6 --- /dev/null +++ b/packages/@n8n/scheduler/vite.config.ts @@ -0,0 +1,14 @@ +import { defineConfig, mergeConfig } from 'vite'; +import { vitestConfig } from '@n8n/vitest-config/node'; +import path from 'node:path'; + +export default mergeConfig( + defineConfig({ + resolve: { + alias: { + '@': path.resolve(__dirname, './src'), + }, + }, + }), + vitestConfig, +); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 310a3fab2ba..cf30fb503a5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -315,6 +315,9 @@ catalogs: cron: specifier: 4.4.0 version: 4.4.0 + cron-parser: + specifier: 5.6.1 + version: 5.6.1 cross-env: specifier: ^7.0.3 version: 7.0.3 @@ -3165,6 +3168,37 @@ importers: specifier: 'catalog:' version: 4.1.9(@opentelemetry/api@1.9.0)(@types/node@20.19.41)(@vitest/browser-playwright@4.1.9)(@vitest/coverage-v8@4.1.9)(jsdom@23.0.1(bufferutil@4.0.9)(utf-8-validate@5.0.10))(vite@8.0.2(@types/node@20.19.41)(esbuild@0.28.1)(jiti@2.6.1)(sass-embedded@1.98.0)(sass@1.98.0)(terser@5.16.1)(tsx@4.19.3)(yaml@2.8.3)) + packages/@n8n/scheduler: + dependencies: + cron-parser: + specifier: 'catalog:' + version: 5.6.1 + luxon: + specifier: 'catalog:' + version: 3.7.2 + n8n-workflow: + specifier: workspace:* + version: link:../../workflow + devDependencies: + '@n8n/typescript-config': + specifier: workspace:* + version: link:../typescript-config + '@n8n/vitest-config': + specifier: workspace:* + version: link:../vitest-config + '@types/luxon': + specifier: 'catalog:' + version: 3.2.0 + '@vitest/coverage-v8': + specifier: 'catalog:' + version: 4.1.9(@vitest/browser@4.1.9)(vitest@4.1.9) + vitest: + specifier: 'catalog:' + version: 4.1.9(@opentelemetry/api@1.9.0)(@types/node@20.19.41)(@vitest/browser-playwright@4.1.9)(@vitest/coverage-v8@4.1.9)(jsdom@23.0.1(bufferutil@4.0.9)(utf-8-validate@5.0.10))(vite@8.0.2(@types/node@20.19.41)(esbuild@0.28.1)(jiti@2.6.1)(sass-embedded@1.98.0)(sass@1.98.0)(terser@5.16.1)(tsx@4.19.3)(yaml@2.8.3)) + vitest-mock-extended: + specifier: 'catalog:' + version: 3.1.0(typescript@6.0.2)(vitest@4.1.9) + packages/@n8n/stylelint-config: dependencies: postcss-html: @@ -14135,6 +14169,10 @@ packages: resolution: {integrity: sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==} engines: {node: '>=12.0.0'} + cron-parser@5.6.1: + resolution: {integrity: sha512-QBm4o1PwZiuY7KFbVvW7FLC8bozy7YWzv+Fz6KRS7sQghzcbDZCGxr/Bc5b6TQreAoSwuWVP491dIcK0THCX6A==} + engines: {node: '>=18'} + cron@4.4.0: resolution: {integrity: sha512-fkdfq+b+AHI4cKdhZlppHveI/mgz2qpiYxcm+t5E5TsxX7QrLS1VE0+7GENEk9z0EeGPcpSciGv6ez24duWhwQ==} engines: {node: '>=18.x'} @@ -32231,6 +32269,10 @@ snapshots: dependencies: luxon: 3.7.2 + cron-parser@5.6.1: + dependencies: + luxon: 3.7.2 + cron@4.4.0: dependencies: '@types/luxon': 3.7.1 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 144f3fe5c97..a54e245730a 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -119,6 +119,7 @@ catalog: cheerio: 1.1.0 chokidar: 4.0.3 cron: 4.4.0 + cron-parser: 5.6.1 cross-env: ^7.0.3 csv-parse: 6.2.1 dotenv: 17.2.3