feat(core): Add durable scheduler retention pruning (no-changelog) (#33616)

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

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

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: {

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,

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

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

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<ScheduledTask> {
 		);
 		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<number> {
+		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<number> {
+		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<number> {
+		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;
+	}
 }

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', () => {

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')`;
 }
 
 /**

packages/@n8n/scheduler/src/core/enums.ts

@@ -7,4 +7,6 @@ export {
 	ScheduledJobKindList,
 	ScheduledTaskStatus,
 	ScheduledTaskStatusList,
+	type TerminalTaskStatus,
+	TerminalTaskStatusList,
 } from '@n8n/db';

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

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

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<number> {
+		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,
+			},
+		]);
+	});
+});

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

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

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<RetentionSummary> {
+	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<PassState> {
+	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 };
+}

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<number>;
+}

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<SchedulerConfig> = {}) {
+	const tasks = mock<ScheduledTaskRepository>();
+	const logger = mock<Logger>();
+	const config = mock<SchedulerConfig>({
+		materializationWindowSeconds: 3600,
+		retentionSeconds: 43_200,
+		failedRetentionSeconds: 86_400,
+		...configOverrides,
+	});
+	const service = new SchedulerService(mock<MaterializerStore>(), 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();
+	});
+});

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<MaterializerSummary> {
-		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<RetentionSummary> {
+		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;
 	}
 }

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<ScheduledTaskEntity> = {},
+	): Promise<ScheduledTaskEntity> {
+		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);
+			},
+		);
+	});
 });

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<ScheduledJob> = {}) =>
+		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<ScheduledTask> = {}) => {
+		// 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);
+	});
+});