From 36c37f44882875f03b14edf298eb0c3b913f154c Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Mon, 6 Jul 2026 10:59:34 +0200 Subject: [PATCH] feat(core): Add durable scheduler retention pruning (no-changelog) (#33616) --- .../__tests__/scheduler.config.test.ts | 12 +- .../config/src/configs/scheduler.config.ts | 40 +++- packages/@n8n/config/test/config.test.ts | 4 +- packages/@n8n/db/src/entities/index.ts | 10 +- .../@n8n/db/src/entities/scheduled-task.ts | 10 + packages/@n8n/db/src/repositories/index.ts | 6 +- .../repositories/scheduled-task.repository.ts | 87 ++++++- .../src/utils/__tests__/dialect-time.test.ts | 20 ++ packages/@n8n/db/src/utils/dialect-time.ts | 9 +- packages/@n8n/scheduler/src/core/enums.ts | 2 + packages/@n8n/scheduler/src/core/errors.ts | 11 + packages/@n8n/scheduler/src/core/index.ts | 20 +- .../core/retention/__tests__/prune.test.ts | 155 ++++++++++++ .../scheduler/src/core/retention/index.ts | 3 + .../scheduler/src/core/retention/options.ts | 38 +++ .../scheduler/src/core/retention/prune.ts | 121 ++++++++++ .../scheduler/src/core/retention/store.ts | 20 ++ .../__tests__/scheduler.service.test.ts | 99 ++++++++ .../src/storage/scheduler.service.ts | 53 ++++- .../scheduling/scheduled-repositories.test.ts | 224 +++++++++++++++++- .../scheduling/scheduler-retention.test.ts | 193 +++++++++++++++ 21 files changed, 1112 insertions(+), 25 deletions(-) create mode 100644 packages/@n8n/scheduler/src/core/retention/__tests__/prune.test.ts create mode 100644 packages/@n8n/scheduler/src/core/retention/index.ts create mode 100644 packages/@n8n/scheduler/src/core/retention/options.ts create mode 100644 packages/@n8n/scheduler/src/core/retention/prune.ts create mode 100644 packages/@n8n/scheduler/src/core/retention/store.ts create mode 100644 packages/@n8n/scheduler/src/storage/__tests__/scheduler.service.test.ts create mode 100644 packages/cli/test/integration/scheduling/scheduler-retention.test.ts diff --git a/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts b/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts index 9c92ab02688..b4ddb5d5490 100644 --- a/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts +++ b/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts @@ -28,7 +28,9 @@ describe('SchedulerConfig', () => { expect(scheduler.claimBatchSize).toBe(100); expect(scheduler.reaperIntervalSeconds).toBe(30); expect(scheduler.leaseDurationSeconds).toBe(60); - expect(scheduler.retentionSeconds).toBe(7 * 24 * 60 * 60); + expect(scheduler.retentionSeconds).toBe(24 * 60 * 60); + expect(scheduler.failedRetentionSeconds).toBe(7 * 24 * 60 * 60); + expect(scheduler.retentionIntervalSeconds).toBe(60 * 60); expect(scheduler.minIntervalSeconds).toBe(0); }); }); @@ -48,7 +50,9 @@ describe('SchedulerConfig', () => { vi.stubEnv('N8N_SCHEDULER_EXECUTOR_INTERVAL', '2'); vi.stubEnv('N8N_SCHEDULER_REAPER_INTERVAL', '45'); vi.stubEnv('N8N_SCHEDULER_LEASE_DURATION', '90'); - vi.stubEnv('N8N_SCHEDULER_RETENTION', '86400'); + vi.stubEnv('N8N_SCHEDULER_RETENTION', '43200'); + vi.stubEnv('N8N_SCHEDULER_FAILED_RETENTION', '86400'); + vi.stubEnv('N8N_SCHEDULER_RETENTION_INTERVAL', '600'); vi.stubEnv('N8N_SCHEDULER_MIN_INTERVAL', '15'); const { scheduler } = Container.get(GlobalConfig); @@ -58,7 +62,9 @@ describe('SchedulerConfig', () => { expect(scheduler.executorIntervalSeconds).toBe(2); expect(scheduler.reaperIntervalSeconds).toBe(45); expect(scheduler.leaseDurationSeconds).toBe(90); - expect(scheduler.retentionSeconds).toBe(86400); + expect(scheduler.retentionSeconds).toBe(43200); + expect(scheduler.failedRetentionSeconds).toBe(86400); + expect(scheduler.retentionIntervalSeconds).toBe(600); expect(scheduler.minIntervalSeconds).toBe(15); }); diff --git a/packages/@n8n/config/src/configs/scheduler.config.ts b/packages/@n8n/config/src/configs/scheduler.config.ts index ba2724c9c21..3eebb7ffae9 100644 --- a/packages/@n8n/config/src/configs/scheduler.config.ts +++ b/packages/@n8n/config/src/configs/scheduler.config.ts @@ -96,14 +96,44 @@ export class SchedulerConfig { leaseDurationSeconds: number = Time.minutes.toSeconds; /** - * How long, in seconds, finished runs are kept in the scheduler's tables before - * being deleted. Defaults to 7 days. + * How long, in seconds, tasks that finished cleanly + * (succeeded or were cancelled) are kept before being deleted. * - * Raise it to keep scheduling history longer for auditing; lower it to reclaim - * database space sooner. Must be greater than 0. + * These rows exist only as recent history, so a short window keeps the + * scheduler's run table small on busy instances. + * + * Raise it to keep scheduling history longer for auditing. + * Lower it to reclaim database space sooner. + * + * Defaults to 1 day. + * Must be greater than 0. */ @Env('N8N_SCHEDULER_RETENTION', positiveIntSchema) - retentionSeconds: number = 7 * Time.days.toSeconds; + retentionSeconds: number = Time.days.toSeconds; + + /** + * How long, in seconds, tasks that went wrong + * (failed, or missed their moment entirely) are kept before being deleted. + * + * Meant to be kept longer than cleanly finished runs (`N8N_SCHEDULER_RETENTION`) + * so there is time to notice and debug a problem before its evidence is deleted. + * + * The scheduler warns when this is set below it. + * Must be greater than 0. + * Defaults to 7 days. + */ + @Env('N8N_SCHEDULER_FAILED_RETENTION', positiveIntSchema) + failedRetentionSeconds: number = 7 * Time.days.toSeconds; + + /** + * How often, in seconds, the scheduler deletes finished tasks older than the + * retention windows above. + * + * Must be greater than 0. + * Defaults to 1 hour. + */ + @Env('N8N_SCHEDULER_RETENTION_INTERVAL', positiveIntSchema) + retentionIntervalSeconds: number = Time.hours.toSeconds; /** * The smallest gap, in seconds, allowed between consecutive runs of the same diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index 524895bab90..5511c109ac9 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -445,7 +445,9 @@ describe('GlobalConfig', () => { claimBatchSize: 100, reaperIntervalSeconds: 30, leaseDurationSeconds: 60, - retentionSeconds: 604800, + retentionSeconds: 86400, + failedRetentionSeconds: 604800, + retentionIntervalSeconds: 3600, minIntervalSeconds: 0, }, evaluation: { diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index 46710a68515..23619ff7f05 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/packages/@n8n/db/src/entities/index.ts @@ -29,7 +29,13 @@ import type { SecretsProviderAccessRole } from './project-secrets-provider-acces import { Role } from './role'; import { RoleMappingRule } from './role-mapping-rule'; import { ScheduledJob, ScheduledJobKind, ScheduledJobKindList } from './scheduled-job'; -import { ScheduledTask, ScheduledTaskStatus, ScheduledTaskStatusList } from './scheduled-task'; +import { + ScheduledTask, + ScheduledTaskStatus, + ScheduledTaskStatusList, + type TerminalTaskStatus, + TerminalTaskStatusList, +} from './scheduled-task'; import { Scope } from './scope'; import { SecretsProviderConnection } from './secrets-provider-connection'; import { Settings } from './settings'; @@ -87,6 +93,8 @@ export { ScheduledTask, ScheduledTaskStatus, ScheduledTaskStatusList, + type TerminalTaskStatus, + TerminalTaskStatusList, Scope, SharedCredentials, SharedWorkflow, diff --git a/packages/@n8n/db/src/entities/scheduled-task.ts b/packages/@n8n/db/src/entities/scheduled-task.ts index b5cd25ce282..5f22c981280 100644 --- a/packages/@n8n/db/src/entities/scheduled-task.ts +++ b/packages/@n8n/db/src/entities/scheduled-task.ts @@ -20,6 +20,16 @@ export type ScheduledTaskStatus = (typeof ScheduledTaskStatus)[keyof typeof Sche /** All statuses as a runtime list. */ export const ScheduledTaskStatusList = Object.values(ScheduledTaskStatus); +/** Statuses of finished work: the only rows retention may delete. */ +export const TerminalTaskStatusList = [ + ScheduledTaskStatus.Succeeded, + ScheduledTaskStatus.Failed, + ScheduledTaskStatus.Missed, + ScheduledTaskStatus.Cancelled, +] as const; + +export type TerminalTaskStatus = (typeof TerminalTaskStatusList)[number]; + /** * One concrete run of a {@link ScheduledJob} at a specific time. * diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index 4e5d3a3b071..6feda385303 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/packages/@n8n/db/src/repositories/index.ts @@ -37,7 +37,11 @@ export { RoleRepository } from './role.repository'; export { RoleMappingRuleRepository } from './role-mapping-rule.repository'; export { ScheduledJobRepository } from './scheduled-job.repository'; export { ScheduledTaskRepository } from './scheduled-task.repository'; -export type { ClaimDueTasksOptions, ClaimRef } from './scheduled-task.repository'; +export type { + ClaimDueTasksOptions, + ClaimRef, + DeleteFinishedTasksOptions, +} from './scheduled-task.repository'; export { ProcessedDataRepository } from './processed-data.repository'; export { SettingsRepository } from './settings.repository'; export { TagRepository } from './tag.repository'; diff --git a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts index e3888686643..526c389334d 100644 --- a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts @@ -4,7 +4,12 @@ import { DataSource, type EntityManager, In, Repository } from '@n8n/typeorm'; import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity'; import { UnexpectedError } from 'n8n-workflow'; -import { ScheduledTask, ScheduledTaskStatus } from '../entities/scheduled-task'; +import { + ScheduledTask, + ScheduledTaskStatus, + type TerminalTaskStatus, + TerminalTaskStatusList, +} from '../entities/scheduled-task'; import { dbNowLiteral, dbNowPlusMsLiteral } from '../utils/dialect-time'; /** Inputs to a claim (see {@link ScheduledTaskRepository.claimDueTasks}). */ @@ -31,6 +36,19 @@ export interface ClaimRef { claimedEpoch: number; } +/** + * Inputs to one retention delete batch + * (see {@link ScheduledTaskRepository.deleteFinishedOlderThan}). + */ +export interface DeleteFinishedTasksOptions { + /** Terminal statuses this batch may delete. Live statuses are rejected. */ + statuses: TerminalTaskStatus[]; + /** Minimum age: only rows whose `finishedAt` is at least this far before DB-now go. */ + olderThanMs: number; + /** Cap on how many rows this one statement deletes. Must be an integer; non-positive is a no-op. */ + limit: number; +} + /** * The columns set when the materializer records an occurrence. */ @@ -276,4 +294,71 @@ export class ScheduledTaskRepository extends Repository { ); return result.affected ?? 0; } + + /** + * Delete up to `limit` finished tasks in `statuses` whose + * `finishedAt` is at least `olderThanMs` before DB-now, oldest first. + * Age is judged against the database clock. + * + * A terminal row missing `finishedAt` (which transitions always set) is skipped. + * + * Concurrent pruners don't fight over rows: + * - Postgres skips locked rows (`FOR UPDATE SKIP LOCKED`), handing simultaneous batches disjoint rows + * - SQLite runs writers one at a time. + * + * @returns how many rows were deleted + * @throws UnexpectedError when `statuses` contains a non-terminal value or `limit` is not an integer + */ + async deleteFinishedOlderThan(options: DeleteFinishedTasksOptions): Promise { + const invalid = options.statuses.filter((status) => !TerminalTaskStatusList.includes(status)); + if (invalid.length > 0) { + throw new UnexpectedError( + `deleteFinishedOlderThan only deletes terminal tasks, got: ${invalid.join(', ')}`, + ); + } + // A non-integer bound to LIMIT errors on SQLite (datatype mismatch), and NaN + // binds as NULL, which on Postgres means LIMIT ALL; reject it before the SQL. + if (!Number.isSafeInteger(options.limit)) { + throw new UnexpectedError( + `deleteFinishedOlderThan needs an integer limit, got: ${options.limit}`, + ); + } + if (options.statuses.length === 0 || options.limit <= 0) { + return 0; + } + return this.isPostgres + ? await this.deleteFinishedWithPostgres(options) + : await this.deleteFinishedWithSqlite(options); + } + + private async deleteFinishedWithPostgres(options: DeleteFinishedTasksOptions): Promise { + const [, affected] = await this.manager.query<[unknown[], number]>( + `DELETE FROM ${this.tableName} + WHERE "id" IN ( + SELECT t."id" FROM ${this.tableName} t + WHERE t."status" = ANY($1) + AND t."finishedAt" <= ${dbNowPlusMsLiteral(true, -options.olderThanMs)} + ORDER BY t."finishedAt" + LIMIT $2 + FOR UPDATE SKIP LOCKED)`, + [options.statuses, options.limit], + ); + return affected; + } + + private async deleteFinishedWithSqlite(options: DeleteFinishedTasksOptions): Promise { + const result = await this.createQueryBuilder() + .delete() + .where( + `id IN ( + SELECT "id" FROM ${this.tableName} + WHERE "status" IN (:...statuses) + AND "finishedAt" <= ${dbNowPlusMsLiteral(false, -options.olderThanMs)} + ORDER BY "finishedAt" + LIMIT :limit)`, + { statuses: options.statuses, limit: options.limit }, + ) + .execute(); + return result.affected ?? 0; + } } diff --git a/packages/@n8n/db/src/utils/__tests__/dialect-time.test.ts b/packages/@n8n/db/src/utils/__tests__/dialect-time.test.ts index 3ec9e5b2e5e..656006b7db4 100644 --- a/packages/@n8n/db/src/utils/__tests__/dialect-time.test.ts +++ b/packages/@n8n/db/src/utils/__tests__/dialect-time.test.ts @@ -25,6 +25,26 @@ describe('dbNowPlusMsLiteral', () => { "CURRENT_TIMESTAMP(3) + (1000 || ' milliseconds')::interval", ); }); + + it('offsets the DB clock into the past on postgres', () => { + expect(dbNowPlusMsLiteral(true, -1500)).toBe( + "CURRENT_TIMESTAMP(3) + (-1500 || ' milliseconds')::interval", + ); + }); + + // A '+-1.5 seconds' modifier would be invalid: STRFTIME returns NULL and the + // comparison silently matches nothing, so the sign must replace the '+'. + it('offsets the DB clock into the past on sqlite without a doubled sign', () => { + expect(dbNowPlusMsLiteral(false, -1500)).toBe( + "STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', '-1.5 seconds')", + ); + }); + + it('treats a negative offset that rounds to zero as now on sqlite', () => { + expect(dbNowPlusMsLiteral(false, -0.4)).toBe( + "STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', '+0 seconds')", + ); + }); }); describe('parseDbTime', () => { diff --git a/packages/@n8n/db/src/utils/dialect-time.ts b/packages/@n8n/db/src/utils/dialect-time.ts index 2e32a7cb677..011f110529b 100644 --- a/packages/@n8n/db/src/utils/dialect-time.ts +++ b/packages/@n8n/db/src/utils/dialect-time.ts @@ -16,13 +16,16 @@ export function dbNowLiteral(isPostgres: boolean): string { /** * DB-clock `now` plus a millisecond offset, per dialect. + * A negative `ms` gives an instant in the past (e.g. a retention cutoff). * `ms` is caller-computed (safe to inline). */ export function dbNowPlusMsLiteral(isPostgres: boolean, ms: number): string { const rounded = Math.round(ms); - return isPostgres - ? `CURRENT_TIMESTAMP(3) + (${rounded} || ' milliseconds')::interval` - : `STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', '+${rounded / 1000} seconds')`; + if (isPostgres) { + return `CURRENT_TIMESTAMP(3) + (${rounded} || ' milliseconds')::interval`; + } + const seconds = rounded / 1000; + return `STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', '${seconds < 0 ? '' : '+'}${seconds} seconds')`; } /** diff --git a/packages/@n8n/scheduler/src/core/enums.ts b/packages/@n8n/scheduler/src/core/enums.ts index 06de7c80972..c1e5d99d8ca 100644 --- a/packages/@n8n/scheduler/src/core/enums.ts +++ b/packages/@n8n/scheduler/src/core/enums.ts @@ -7,4 +7,6 @@ export { ScheduledJobKindList, ScheduledTaskStatus, ScheduledTaskStatusList, + type TerminalTaskStatus, + TerminalTaskStatusList, } from '@n8n/db'; diff --git a/packages/@n8n/scheduler/src/core/errors.ts b/packages/@n8n/scheduler/src/core/errors.ts index 72885df4044..bbf2e77e9d0 100644 --- a/packages/@n8n/scheduler/src/core/errors.ts +++ b/packages/@n8n/scheduler/src/core/errors.ts @@ -13,6 +13,17 @@ export class InvalidScheduleError extends Error { } } +/** + * Raised when a retention pass is invoked with unusable options (e.g. a + * non-positive batch size), before any delete statement is issued. + */ +export class InvalidRetentionOptionsError extends Error { + constructor(message: string) { + super(message); + this.name = 'InvalidRetentionOptionsError'; + } +} + /** * Raised when a stored row is missing a column its `kind`/type guarantees * should be set (a corrupt or hand-edited row), while mapping DB entities to diff --git a/packages/@n8n/scheduler/src/core/index.ts b/packages/@n8n/scheduler/src/core/index.ts index 02038a8686f..dacd5ef52a9 100644 --- a/packages/@n8n/scheduler/src/core/index.ts +++ b/packages/@n8n/scheduler/src/core/index.ts @@ -11,6 +11,10 @@ * / interval / one-off `Schedule` is materialized into tasks by the materializer * (see `materializer/`). All time and DST math is confined to this boundary * (see `recurrence/`). + * + * *Retention* (see `retention/`) bounds the task table: terminal tasks past + * their window are deleted in bounded batches, so the queue's history cannot + * grow without bound. */ // These enums live in `@n8n/db` (the schema is their source of truth), re-exported @@ -20,6 +24,8 @@ export { ScheduledJobKindList, ScheduledTaskStatus, ScheduledTaskStatusList, + type TerminalTaskStatus, + TerminalTaskStatusList, } from './enums'; export type { @@ -31,7 +37,11 @@ export type { ScheduledTask, } from './types'; -export { InvalidScheduleError, CorruptStorageRowError } from './errors'; +export { + InvalidScheduleError, + InvalidRetentionOptionsError, + CorruptStorageRowError, +} from './errors'; export { computeNextRunAt } from './recurrence/next-run'; export { validateSchedule } from './recurrence/validate'; @@ -49,3 +59,11 @@ export type { RunInTransaction, MaterializerTransaction, } from './materializer'; + +export { prune, DEFAULT_RETENTION_OPTIONS } from './retention'; +export type { + RetentionSummary, + RetentionOptions, + RetentionBatch, + RetentionStore, +} from './retention'; diff --git a/packages/@n8n/scheduler/src/core/retention/__tests__/prune.test.ts b/packages/@n8n/scheduler/src/core/retention/__tests__/prune.test.ts new file mode 100644 index 00000000000..d457923068e --- /dev/null +++ b/packages/@n8n/scheduler/src/core/retention/__tests__/prune.test.ts @@ -0,0 +1,155 @@ +import { ScheduledTaskStatus } from '../../enums'; +import { DEFAULT_RETENTION_OPTIONS, type RetentionOptions } from '../options'; +import { prune } from '../prune'; +import type { RetentionBatch, RetentionStore } from '../store'; + +const CLEAN = [ScheduledTaskStatus.Succeeded, ScheduledTaskStatus.Cancelled]; +const WRONG = [ScheduledTaskStatus.Failed, ScheduledTaskStatus.Missed]; + +/** A store that records every batch and replays scripted per-call row counts (0 once exhausted). */ +class RecordingStore implements RetentionStore { + readonly batches: RetentionBatch[] = []; + + constructor(private readonly results: number[] = []) {} + + async deleteFinishedOlderThan(batch: RetentionBatch): Promise { + this.batches.push(batch); + return await Promise.resolve(this.results[this.batches.length - 1] ?? 0); + } +} + +const options: RetentionOptions = { + retentionSeconds: 60, + failedRetentionSeconds: 3600, + batchSize: 10, + maxBatchesPerPass: 100, +}; + +describe('prune', () => { + it('prunes each window with its own statuses and cutoff', async () => { + const store = new RecordingStore(); + + const summary = await prune(store, options); + + expect(summary).toEqual({ deleted: 0, drained: true }); + expect(store.batches).toEqual([ + { statuses: CLEAN, olderThanMs: 60_000, limit: 10 }, + { statuses: WRONG, olderThanMs: 3_600_000, limit: 10 }, + ]); + }); + + it('keeps deleting a window until a batch comes back short', async () => { + // Two full batches then a short one drain the clean window; the second + // window is empty on its first probe. + const store = new RecordingStore([10, 10, 3, 0]); + + const summary = await prune(store, options); + + expect(summary).toEqual({ deleted: 23, drained: true }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, CLEAN, CLEAN, WRONG]); + }); + + it('spends one extra batch to prove a boundary-exact window is empty', async () => { + // The first batch is full by coincidence (exactly 10 rows were eligible), + // so only the following empty batch ends the window's drain. + const store = new RecordingStore([10, 0, 0]); + + const summary = await prune(store, options); + + expect(summary).toEqual({ deleted: 10, drained: true }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, CLEAN, WRONG]); + }); + + it('stops at the pass budget and reports the pass undrained', async () => { + // Every batch full: the backlog outlives the budget. The second window's + // reserved probe still runs, so a saturated first window can't starve it. + const store = new RecordingStore([10, 10, 10, 10]); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 2 }); + + expect(summary).toEqual({ deleted: 20, drained: false }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, WRONG]); + }); + + it('probes the second window when the first drains on the last unreserved batch', async () => { + // Window 1's full batch spends all the budget it may take; the reserved + // statement still proves window 2's state instead of skipping it. + const store = new RecordingStore([10, 3]); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 2 }); + + expect(summary).toEqual({ deleted: 13, drained: false }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, WRONG]); + }); + + it('proves an empty table drained with exactly one probe per window', async () => { + const store = new RecordingStore(); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 2 }); + + expect(summary).toEqual({ deleted: 0, drained: true }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, WRONG]); + }); + + it('cannot prove both windows drained with a single-statement budget', async () => { + // One statement can only ever probe the first window; the second stays + // unprobed, so the pass is honestly reported undrained. + const store = new RecordingStore(); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 1 }); + + expect(summary).toEqual({ deleted: 0, drained: false }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN]); + }); + + it.each([0, -1, 1.5, NaN])( + 'rejects batch size %p before issuing any statement', + async (batchSize) => { + const store = new RecordingStore([10]); + + await expect(prune(store, { ...options, batchSize })).rejects.toThrow( + 'batchSize must be a positive integer', + ); + expect(store.batches).toHaveLength(0); + }, + ); + + it('shares one budget across both windows', async () => { + // The clean window drains in two batches, leaving one for the other + // window; its full batch exhausts the budget before proving emptiness. + const store = new RecordingStore([10, 4, 10]); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 3 }); + + expect(summary).toEqual({ deleted: 24, drained: false }); + expect(store.batches.map((batch) => batch.statuses)).toEqual([CLEAN, CLEAN, WRONG]); + }); + + it('deletes nothing and reports undrained when the budget is zero', async () => { + const store = new RecordingStore([10]); + + const summary = await prune(store, { ...options, maxBatchesPerPass: 0 }); + + expect(summary).toEqual({ deleted: 0, drained: false }); + expect(store.batches).toHaveLength(0); + }); + + it('falls back to the default windows and batch bounds', async () => { + const store = new RecordingStore(); + + await prune(store); + + expect(store.batches).toEqual([ + { + statuses: CLEAN, + olderThanMs: DEFAULT_RETENTION_OPTIONS.retentionSeconds * 1000, + limit: DEFAULT_RETENTION_OPTIONS.batchSize, + }, + { + statuses: WRONG, + olderThanMs: DEFAULT_RETENTION_OPTIONS.failedRetentionSeconds * 1000, + limit: DEFAULT_RETENTION_OPTIONS.batchSize, + }, + ]); + }); +}); diff --git a/packages/@n8n/scheduler/src/core/retention/index.ts b/packages/@n8n/scheduler/src/core/retention/index.ts new file mode 100644 index 00000000000..796128c1e03 --- /dev/null +++ b/packages/@n8n/scheduler/src/core/retention/index.ts @@ -0,0 +1,3 @@ +export { prune, type RetentionSummary } from './prune'; +export { DEFAULT_RETENTION_OPTIONS, type RetentionOptions } from './options'; +export type { RetentionBatch, RetentionStore } from './store'; diff --git a/packages/@n8n/scheduler/src/core/retention/options.ts b/packages/@n8n/scheduler/src/core/retention/options.ts new file mode 100644 index 00000000000..3c4fcc8e7c1 --- /dev/null +++ b/packages/@n8n/scheduler/src/core/retention/options.ts @@ -0,0 +1,38 @@ +import { Time } from '@n8n/constants'; + +/** + * Knobs of a retention pass. + * The trade-offs are documented on `prune`. + */ +export interface RetentionOptions { + /** + * How long tasks that finished cleanly (succeeded, cancelled) are kept after finishing, in seconds. + * Must be > 0. + */ + retentionSeconds: number; + + /** + * How long tasks that went wrong (failed, missed) are kept after finishing, in seconds. + * Must be > 0. + */ + failedRetentionSeconds: number; + + /** + * The most rows one delete statement removes. + * Must be a positive integer. + */ + batchSize: number; + + /** + * The most delete statements one pass issues, shared across both windows, + * bounding the pass; a backlog beyond it drains over successive passes. + */ + maxBatchesPerPass: number; +} + +export const DEFAULT_RETENTION_OPTIONS: RetentionOptions = { + retentionSeconds: Time.days.toSeconds, + failedRetentionSeconds: 7 * Time.days.toSeconds, + batchSize: 1000, + maxBatchesPerPass: 1000, +}; diff --git a/packages/@n8n/scheduler/src/core/retention/prune.ts b/packages/@n8n/scheduler/src/core/retention/prune.ts new file mode 100644 index 00000000000..44fbdb91a61 --- /dev/null +++ b/packages/@n8n/scheduler/src/core/retention/prune.ts @@ -0,0 +1,121 @@ +import { Time } from '@n8n/constants'; + +import { ScheduledTaskStatus, type TerminalTaskStatus } from '../enums'; +import { InvalidRetentionOptionsError } from '../errors'; +import { DEFAULT_RETENTION_OPTIONS, type RetentionOptions } from './options'; +import type { RetentionStore } from './store'; + +export interface RetentionSummary { + /** How many finished tasks this pass deleted. */ + deleted: number; + + /** + * Whether everything past its window went. + * `false` means the pass spent its batch budget first. + * The next pass continues where this one stopped. + */ + drained: boolean; +} + +interface RetentionWindow { + statuses: TerminalTaskStatus[]; + olderThanSeconds: number; +} + +function retentionWindows(options: RetentionOptions): RetentionWindow[] { + return [ + { + statuses: [ScheduledTaskStatus.Succeeded, ScheduledTaskStatus.Cancelled], + olderThanSeconds: options.retentionSeconds, + }, + { + statuses: [ScheduledTaskStatus.Failed, ScheduledTaskStatus.Missed], + olderThanSeconds: options.failedRetentionSeconds, + }, + ]; +} + +/** + * One retention pass of the scheduler: delete terminal tasks past their + * retention window, oldest first, in bounded batches. + * + * Each batch is one bounded statement, deliberately not wrapped in an + * enclosing transaction: batches don't need atomicity with each other (a + * terminal row never becomes live again, so whatever a partial pass leaves + * behind is simply picked up later), and small statements keep locks short. + * + * A pass issues at most `maxBatchesPerPass` statements, + * so a large backlog drains across successive passes instead of monopolising one. + * + * Each window ahead keeps one statement of that budget in reserve, so a backlog saturating an + * early window cannot starve a later one out of the pass entirely. + */ +export async function prune( + store: RetentionStore, + options: RetentionOptions = DEFAULT_RETENTION_OPTIONS, +): Promise { + if (!Number.isInteger(options.batchSize) || options.batchSize <= 0) { + throw new InvalidRetentionOptionsError( + `batchSize must be a positive integer, got ${options.batchSize}`, + ); + } + + // Only the summary leaves; the budget is pass-internal bookkeeping. + const { deleted, drained } = await pruneWindows(store, retentionWindows(options), options, { + deleted: 0, + budget: options.maxBatchesPerPass, + drained: true, + }); + return { deleted, drained }; +} + +interface PassState { + deleted: number; + budget: number; + drained: boolean; +} + +async function pruneWindows( + store: RetentionStore, + windows: RetentionWindow[], + options: RetentionOptions, + state: PassState, +): Promise { + const [window, ...rest] = windows; + if (window === undefined) { + return state; + } + + const windowBudget = state.budget <= 0 ? 0 : Math.max(state.budget - rest.length, 1); + const pruned = await pruneWindow(store, window, options.batchSize, windowBudget); + return await pruneWindows(store, rest, options, { + deleted: state.deleted + pruned.deleted, + budget: state.budget - pruned.batches, + drained: state.drained && pruned.drained, + }); +} + +async function pruneWindow( + store: RetentionStore, + window: RetentionWindow, + batchSize: number, + budget: number, +): Promise<{ deleted: number; batches: number; drained: boolean }> { + let deleted = 0; + let batches = 0; + + while (batches < budget) { + const affected = await store.deleteFinishedOlderThan({ + statuses: window.statuses, + olderThanMs: window.olderThanSeconds * Time.seconds.toMilliseconds, + limit: batchSize, + }); + batches += 1; + deleted += affected; + if (affected < batchSize) { + return { deleted, batches, drained: true }; + } + } + + return { deleted, batches, drained: false }; +} diff --git a/packages/@n8n/scheduler/src/core/retention/store.ts b/packages/@n8n/scheduler/src/core/retention/store.ts new file mode 100644 index 00000000000..dc303e01036 --- /dev/null +++ b/packages/@n8n/scheduler/src/core/retention/store.ts @@ -0,0 +1,20 @@ +import type { TerminalTaskStatus } from '../enums'; + +export interface RetentionBatch { + /** Terminal statuses this batch may delete. */ + statuses: TerminalTaskStatus[]; + /** Minimum age: only rows finished at least this long before DB-now go. */ + olderThanMs: number; + /** Cap on how many rows this one statement deletes. Non-positive is a no-op. */ + limit: number; +} + +export interface RetentionStore { + /** + * Delete up to `limit` tasks in `statuses` whose `finishedAt` is at least older than `olderThanMs`. + * Oldest first. + * + * @returns how many rows were deleted + */ + deleteFinishedOlderThan(batch: RetentionBatch): Promise; +} diff --git a/packages/@n8n/scheduler/src/storage/__tests__/scheduler.service.test.ts b/packages/@n8n/scheduler/src/storage/__tests__/scheduler.service.test.ts new file mode 100644 index 00000000000..5121c4aa937 --- /dev/null +++ b/packages/@n8n/scheduler/src/storage/__tests__/scheduler.service.test.ts @@ -0,0 +1,99 @@ +import type { Logger } from '@n8n/backend-common'; +import type { SchedulerConfig } from '@n8n/config'; +import type { ScheduledTaskRepository } from '@n8n/db'; +import { mock } from 'vitest-mock-extended'; + +import { ScheduledTaskStatus } from '../../core/enums'; +import { DEFAULT_RETENTION_OPTIONS } from '../../core/retention'; +import type { MaterializerStore } from '../materializer-store'; +import { SchedulerService } from '../scheduler.service'; + +/** Build the service with non-default retention config, returning its mocks. */ +function makeService(configOverrides: Partial = {}) { + const tasks = mock(); + const logger = mock(); + const config = mock({ + materializationWindowSeconds: 3600, + retentionSeconds: 43_200, + failedRetentionSeconds: 86_400, + ...configOverrides, + }); + const service = new SchedulerService(mock(), tasks, logger, config); + return { service, tasks, logger }; +} + +describe('SchedulerService.prune', () => { + it('maps the configured windows into the batches the repository receives', async () => { + const { service, tasks } = makeService(); + tasks.deleteFinishedOlderThan.mockResolvedValue(0); + + const summary = await service.prune(); + + expect(summary).toEqual({ deleted: 0, drained: true }); + expect(tasks.deleteFinishedOlderThan).toHaveBeenNthCalledWith(1, { + statuses: [ScheduledTaskStatus.Succeeded, ScheduledTaskStatus.Cancelled], + olderThanMs: 43_200_000, + limit: DEFAULT_RETENTION_OPTIONS.batchSize, + }); + expect(tasks.deleteFinishedOlderThan).toHaveBeenNthCalledWith(2, { + statuses: [ScheduledTaskStatus.Failed, ScheduledTaskStatus.Missed], + olderThanMs: 86_400_000, + limit: DEFAULT_RETENTION_OPTIONS.batchSize, + }); + }); + + it('warns when a pass spends its batch budget with backlog remaining', async () => { + const { service, tasks, logger } = makeService(); + // Every batch full: the pass can never prove either window drained. + tasks.deleteFinishedOlderThan.mockResolvedValue(DEFAULT_RETENTION_OPTIONS.batchSize); + + const summary = await service.prune(); + + expect(summary.drained).toBe(false); + expect(logger.warn).toHaveBeenCalledWith( + 'Scheduler retention pass hit its batch budget; backlog remains', + { ...summary }, + ); + expect(logger.debug).not.toHaveBeenCalled(); + }); + + it('logs a drained pass that deleted rows at debug only', async () => { + const { service, tasks, logger } = makeService(); + tasks.deleteFinishedOlderThan.mockResolvedValueOnce(5).mockResolvedValue(0); + + const summary = await service.prune(); + + expect(summary).toEqual({ deleted: 5, drained: true }); + expect(logger.debug).toHaveBeenCalledWith('Scheduler retention deleted finished tasks', { + ...summary, + }); + expect(logger.warn).not.toHaveBeenCalled(); + }); + + it('logs nothing on a no-op pass', async () => { + const { service, tasks, logger } = makeService(); + tasks.deleteFinishedOlderThan.mockResolvedValue(0); + + await service.prune(); + + expect(logger.debug).not.toHaveBeenCalled(); + expect(logger.warn).not.toHaveBeenCalled(); + }); +}); + +describe('SchedulerService retention config', () => { + it('warns at construction when failed runs are kept shorter than clean ones', () => { + const { logger } = makeService({ retentionSeconds: 86_400, failedRetentionSeconds: 3600 }); + + expect(logger.warn).toHaveBeenCalledWith( + 'Scheduler retention keeps failed runs shorter than succeeded ones; failure evidence will be deleted first', + { retentionSeconds: 86_400, failedRetentionSeconds: 3600 }, + ); + }); + + it('stays silent when failed runs are kept at least as long', () => { + const { logger } = makeService(); + + expect(logger.warn).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/@n8n/scheduler/src/storage/scheduler.service.ts b/packages/@n8n/scheduler/src/storage/scheduler.service.ts index 433675d7405..fba58f144d4 100644 --- a/packages/@n8n/scheduler/src/storage/scheduler.service.ts +++ b/packages/@n8n/scheduler/src/storage/scheduler.service.ts @@ -1,10 +1,13 @@ import { Logger } from '@n8n/backend-common'; import { SchedulerConfig } from '@n8n/config'; +import { ScheduledTaskRepository } from '@n8n/db'; import { Service } from '@n8n/di'; import { MaterializerStore } from './materializer-store'; import { DEFAULT_MATERIALIZER_OPTIONS, materialize } from '../core/materializer'; import type { MaterializerOptions, MaterializerSummary } from '../core/materializer'; +import { DEFAULT_RETENTION_OPTIONS, prune } from '../core/retention'; +import type { RetentionOptions, RetentionSummary } from '../core/retention'; /** * This is the entry the lifecycle wiring drives on a timer; it does not schedule itself. @@ -12,25 +15,61 @@ import type { MaterializerOptions, MaterializerSummary } from '../core/materiali */ @Service() export class SchedulerService { - private readonly options: MaterializerOptions; + private readonly materializerOptions: MaterializerOptions; + + private readonly retentionOptions: RetentionOptions; constructor( private readonly store: MaterializerStore, + private readonly tasks: ScheduledTaskRepository, private readonly logger: Logger, config: SchedulerConfig, ) { - this.options = { + this.materializerOptions = { ...DEFAULT_MATERIALIZER_OPTIONS, windowSeconds: config.materializationWindowSeconds, }; + this.retentionOptions = { + ...DEFAULT_RETENTION_OPTIONS, + retentionSeconds: config.retentionSeconds, + failedRetentionSeconds: config.failedRetentionSeconds, + }; + if (config.failedRetentionSeconds < config.retentionSeconds) { + this.logger.warn( + 'Scheduler retention keeps failed runs shorter than succeeded ones; failure evidence will be deleted first', + { + retentionSeconds: config.retentionSeconds, + failedRetentionSeconds: config.failedRetentionSeconds, + }, + ); + } } async materialize(): Promise { - return await materialize(this.store.runInTransaction, this.options, (job, error) => { - this.logger.error('Scheduler could not plan a job schedule; deferred for retry', { - jobId: job.id, - error: error instanceof Error ? error.message : String(error), + return await materialize( + this.store.runInTransaction, + this.materializerOptions, + (job, error) => { + this.logger.error('Scheduler could not plan a job schedule; deferred for retry', { + jobId: job.id, + error: error instanceof Error ? error.message : String(error), + }); + }, + ); + } + + /** + * One retention pass: delete finished tasks past their windows. + */ + async prune(): Promise { + const summary = await prune(this.tasks, this.retentionOptions); + if (!summary.drained) { + this.logger.warn('Scheduler retention pass hit its batch budget; backlog remains', { + ...summary, }); - }); + } else if (summary.deleted > 0) { + this.logger.debug('Scheduler retention deleted finished tasks', { ...summary }); + } + return summary; } } diff --git a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts index 73654ac3519..a8d88fd2489 100644 --- a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts +++ b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts @@ -1,6 +1,15 @@ import { testDb } from '@n8n/backend-test-utils'; -import type { ScheduledJob as ScheduledJobEntity } from '@n8n/db'; -import { DbConnectionOptions, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db'; +import type { + ScheduledJob as ScheduledJobEntity, + ScheduledTask as ScheduledTaskEntity, + TerminalTaskStatus, +} from '@n8n/db'; +import { + DbConnectionOptions, + ScheduledJobRepository, + ScheduledTask, + ScheduledTaskRepository, +} from '@n8n/db'; import { Container } from '@n8n/di'; import { DataSource } from '@n8n/typeorm'; @@ -72,6 +81,26 @@ describe('scheduled repositories', () => { ); } + /** Insert a task in a given lifecycle state; `scheduledFor` is made unique per row. */ + let taskSequence = 0; + async function createTask( + jobId: number, + overrides: Partial = {}, + ): Promise { + const scheduledFor = secondsFromNow(-++taskSequence); + return await taskRepository.save( + taskRepository.create({ + jobId, + taskType: 'scheduleTrigger', + payload: {}, + scheduledFor, + runAt: scheduledFor, + maxAttempts: 1, + ...overrides, + }), + ); + } + describe('transaction', () => { it('commits repository calls made through the transaction manager', async () => { const job = await createJob({ nextRunAt: secondsFromNow(-60) }); @@ -415,4 +444,195 @@ describe('scheduled repositories', () => { }, ); }); + + describe('ScheduledTaskRepository.deleteFinishedOlderThan', () => { + const HOUR_MS = 60 * 60 * 1000; + + it('deletes only rows in the given statuses finished before the cutoff', async () => { + const job = await createJob(); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-7200) }); + await createTask(job.id, { status: 'cancelled', finishedAt: secondsFromNow(-7200) }); + const freshSucceeded = await createTask(job.id, { + status: 'succeeded', + finishedAt: secondsFromNow(-60), + }); + const oldFailed = await createTask(job.id, { + status: 'failed', + finishedAt: secondsFromNow(-7200), + }); + const pending = await createTask(job.id, { status: 'pending' }); + const running = await createTask(job.id, { + status: 'running', + claimedBy: 'main-1', + leaseExpiresAt: secondsFromNow(-7200), + }); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded', 'cancelled'], + olderThanMs: HOUR_MS, + limit: 100, + }); + + // Both expired clean rows went; the fresh one, other statuses, and + // live rows (no finishedAt) survived. + expect(deleted).toBe(2); + const survivors = new Set((await taskRepository.find()).map((t) => t.id)); + expect(survivors).toEqual(new Set([freshSucceeded.id, oldFailed.id, pending.id, running.id])); + }); + + it('deletes the oldest rows first when the limit caps a batch', async () => { + const job = await createJob(); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-4 * 3600) }); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-3 * 3600) }); + const youngest = await createTask(job.id, { + status: 'succeeded', + finishedAt: secondsFromNow(-2 * 3600), + }); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: HOUR_MS, + limit: 2, + }); + + expect(deleted).toBe(2); + const remaining = await taskRepository.find(); + expect(remaining.map((t) => t.id)).toEqual([youngest.id]); + }); + + it('never deletes a terminal row missing finishedAt', async () => { + const job = await createJob(); + // Transitions always set finishedAt; a row without it has no provable + // age, so retention must leave it alone rather than guess. + const untimed = await createTask(job.id, { status: 'succeeded', finishedAt: null }); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: 0, + limit: 100, + }); + + expect(deleted).toBe(0); + expect((await taskRepository.find()).map((t) => t.id)).toEqual([untimed.id]); + }); + + it('deletes nothing when no statuses are given', async () => { + const job = await createJob(); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-7200) }); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: [], + olderThanMs: HOUR_MS, + limit: 100, + }); + + expect(deleted).toBe(0); + expect(await taskRepository.count()).toBe(1); + }); + + it('rejects live statuses before touching any row', async () => { + const job = await createJob(); + const oldPending = await createTask(job.id, { status: 'pending' }); + + // The type already forbids this; the cast simulates a value smuggled + // past it (an untyped caller), which the runtime guard must stop. + await expect( + taskRepository.deleteFinishedOlderThan({ + statuses: ['pending' as TerminalTaskStatus, 'succeeded'], + olderThanMs: 0, + limit: 100, + }), + ).rejects.toThrow('only deletes terminal tasks, got: pending'); + expect((await taskRepository.find()).map((t) => t.id)).toEqual([oldPending.id]); + }); + + it('rejects a non-integer limit before touching any row', async () => { + const job = await createJob(); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-7200) }); + + await expect( + taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: HOUR_MS, + limit: 1.5, + }), + ).rejects.toThrow('needs an integer limit, got: 1.5'); + expect(await taskRepository.count()).toBe(1); + }); + + it('treats a non-positive limit as a no-op batch', async () => { + const job = await createJob(); + await createTask(job.id, { status: 'succeeded', finishedAt: secondsFromNow(-7200) }); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: HOUR_MS, + limit: 0, + }); + + expect(deleted).toBe(0); + expect(await taskRepository.count()).toBe(1); + }); + + // Postgres only: while another transaction holds eligible rows locked, + // FOR UPDATE SKIP LOCKED must hand the concurrent batch the remaining rows + // instead of blocking on (or double-deleting) the locked ones. On sqlite + // the single-writer lock serializes the two, so there is nothing to skip. + it.skipIf(!isPostgres)( + 'skips rows another transaction holds locked and deletes the rest', + async () => { + const job = await createJob(); + const lockedOlder = await createTask(job.id, { + status: 'succeeded', + finishedAt: secondsFromNow(-4 * 3600), + }); + const lockedNewer = await createTask(job.id, { + status: 'succeeded', + finishedAt: secondsFromNow(-3 * 3600), + }); + await createTask(job.id, { + status: 'succeeded', + finishedAt: secondsFromNow(-2 * 3600), + }); + + // The holder draws from the secondary pool so its transaction stays + // open while the delete runs on the main pool. + const runner = secondaryDataSource!.createQueryRunner(); + try { + await runner.connect(); + await runner.startTransaction(); + await runner.manager + .getRepository(ScheduledTask) + .createQueryBuilder('task') + .setLock('pessimistic_write') + .whereInIds([lockedOlder.id, lockedNewer.id]) + .getMany(); + + const deleted = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: HOUR_MS, + limit: 10, + }); + + // Only the unlocked row went, even though the locked ones are older. + expect(deleted).toBe(1); + const survivors = new Set((await taskRepository.find()).map((t) => t.id)); + expect(survivors).toEqual(new Set([lockedOlder.id, lockedNewer.id])); + + await runner.rollbackTransaction(); + } finally { + await runner.release(); + } + + // With the lock released, the next batch reaps what was skipped. + const rest = await taskRepository.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: HOUR_MS, + limit: 10, + }); + expect(rest).toBe(2); + expect(await taskRepository.count()).toBe(0); + }, + ); + }); }); diff --git a/packages/cli/test/integration/scheduling/scheduler-retention.test.ts b/packages/cli/test/integration/scheduling/scheduler-retention.test.ts new file mode 100644 index 00000000000..603ef57f38a --- /dev/null +++ b/packages/cli/test/integration/scheduling/scheduler-retention.test.ts @@ -0,0 +1,193 @@ +import { testDb } from '@n8n/backend-test-utils'; +import type { ScheduledJob, ScheduledTask } from '@n8n/db'; +import { ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { prune } from '@n8n/scheduler'; +import { SchedulerService } from '@n8n/scheduler/storage'; + +describe('scheduler retention', () => { + let jobRepo: ScheduledJobRepository; + let taskRepo: ScheduledTaskRepository; + + beforeAll(async () => { + await testDb.init(); + jobRepo = Container.get(ScheduledJobRepository); + taskRepo = Container.get(ScheduledTaskRepository); + }); + + beforeEach(async () => { + await testDb.truncate(['ScheduledTask', 'ScheduledJob']); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + let sequence = 0; + const daysAgo = (days: number) => new Date(Date.now() - days * 24 * 3600 * 1000); + + const createJob = async (overrides: Partial = {}) => + await jobRepo.save( + jobRepo.create({ + name: `job-${++sequence}`, + taskType: 'test', + payload: {}, + kind: 'interval', + intervalSeconds: 3600, + enabled: true, + nextRunAt: daysAgo(0), + maxAttempts: 1, + ...overrides, + }), + ); + + const createTask = async (jobId: number, overrides: Partial = {}) => { + // Each row gets its own scheduledFor to satisfy the (jobId, scheduledFor) identity. + const scheduledFor = new Date(Date.now() - ++sequence * 1000); + return await taskRepo.save( + taskRepo.create({ + jobId, + taskType: 'test', + payload: {}, + scheduledFor, + runAt: scheduledFor, + maxAttempts: 1, + ...overrides, + }), + ); + }; + + it('prunes each terminal class on its own window through SchedulerService', async () => { + const job = await createJob(); + // Config defaults: cleanly finished kept 1 day, failed/missed kept 7 days. + await createTask(job.id, { status: 'succeeded', finishedAt: daysAgo(2) }); + await createTask(job.id, { status: 'cancelled', finishedAt: daysAgo(2) }); + await createTask(job.id, { status: 'failed', finishedAt: daysAgo(8) }); + await createTask(job.id, { status: 'missed', finishedAt: daysAgo(8) }); + const freshSucceeded = await createTask(job.id, { + status: 'succeeded', + finishedAt: daysAgo(0.5), + }); + // Failed two days ago: past the clean window but within the failed one, + // so it stays around for debugging. + const recentFailed = await createTask(job.id, { status: 'failed', finishedAt: daysAgo(2) }); + const pending = await createTask(job.id, { status: 'pending' }); + const running = await createTask(job.id, { + status: 'running', + claimedBy: 'main-1', + leaseExpiresAt: daysAgo(2), + }); + + const summary = await Container.get(SchedulerService).prune(); + + expect(summary).toEqual({ deleted: 4, drained: true }); + const survivors = new Set((await taskRepo.find()).map((t) => t.id)); + expect(survivors).toEqual( + new Set([freshSucceeded.id, recentFailed.id, pending.id, running.id]), + ); + }); + + it('drains a backlog across passes, the repository serving as the store directly', async () => { + const job = await createJob(); + for (let i = 0; i < 5; i++) { + await createTask(job.id, { status: 'succeeded', finishedAt: daysAgo(2) }); + } + const options = { + retentionSeconds: 24 * 3600, + failedRetentionSeconds: 7 * 24 * 3600, + batchSize: 2, + maxBatchesPerPass: 1, + }; + + // One batch per pass: the first pass deletes batchSize rows and reports + // the backlog it left behind. + const first = await prune(taskRepo, options); + expect(first).toEqual({ deleted: 2, drained: false }); + + // Successive passes keep draining until a pass proves nothing is left. + let deleted = first.deleted; + for (let pass = 0; pass < 10 && deleted < 5; pass++) { + const summary = await prune(taskRepo, options); + deleted += summary.deleted; + } + + expect(deleted).toBe(5); + expect(await taskRepo.count()).toBe(0); + }); + + it('carries the pass budget across both windows against a real database', async () => { + const job = await createJob(); + for (let i = 0; i < 3; i++) { + await createTask(job.id, { status: 'succeeded', finishedAt: daysAgo(2) }); + } + for (let i = 0; i < 2; i++) { + await createTask(job.id, { status: 'failed', finishedAt: daysAgo(8) }); + } + const options = { + retentionSeconds: 24 * 3600, + failedRetentionSeconds: 7 * 24 * 3600, + batchSize: 2, + maxBatchesPerPass: 3, + }; + + // The clean window drains in two batches (2 rows, then a short 1); the + // failed window's reserved statement deletes a full batch, which spends + // the budget without proving that window empty. + const first = await prune(taskRepo, options); + expect(first).toEqual({ deleted: 5, drained: false }); + expect(await taskRepo.count()).toBe(0); + + // The next pass probes both windows and proves the drain. + const second = await prune(taskRepo, options); + expect(second).toEqual({ deleted: 0, drained: true }); + }); + + it('reports a no-op pass as drained with nothing deleted', async () => { + const job = await createJob(); + const fresh = await createTask(job.id, { status: 'succeeded', finishedAt: daysAgo(0.5) }); + + const summary = await Container.get(SchedulerService).prune(); + + expect(summary).toEqual({ deleted: 0, drained: true }); + expect((await taskRepo.find()).map((t) => t.id)).toEqual([fresh.id]); + }); + + it('prunes what the executor completes: the terminal write stamps the age retention reads', async () => { + const job = await createJob(); + await createTask(job.id); + + // The executor's lifecycle against the same rows retention prunes: claim + // the due task, then record its success — completeTask stamps finishedAt + // with the DB clock, and that instant is what the cutoff compares against. + const [claimed] = await taskRepo.claimDueTasks({ + host: 'main-1', + taskTypes: ['test'], + lookaheadMs: 0, + leaseMs: 60_000, + batchSize: 1, + }); + expect(claimed).toBeDefined(); + await taskRepo.completeTask({ + host: 'main-1', + id: claimed.id, + claimedEpoch: claimed.leaseEpoch, + }); + + // Younger than a real window: kept. + const kept = await taskRepo.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: 3_600_000, + limit: 10, + }); + expect(kept).toBe(0); + + // A zero cutoff ages out everything finished: the completed row goes. + const pruned = await taskRepo.deleteFinishedOlderThan({ + statuses: ['succeeded'], + olderThanMs: 0, + limit: 10, + }); + expect(pruned).toBe(1); + expect(await taskRepo.count()).toBe(0); + }); +});