fix(core): Claim scheduler jobs ahead of due to remove materializer window-boundary lag (no-changelog) (#34331)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

packages/@n8n/db/src/repositories/scheduled-job.repository.ts

@@ -6,7 +6,7 @@ import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPar
 import { UnexpectedError } from 'n8n-workflow';
 
 import { ScheduledJob } from '../entities/scheduled-job';
-import { dbNowLiteral, parseDbTime } from '../utils/dialect-time';
+import { dbNowLiteral, dbNowPlusMsLiteral, parseDbTime } from '../utils/dialect-time';
 
 /** The new clock values for one advanced job. */
 export interface JobAdvance {
@@ -63,21 +63,25 @@ export class ScheduledJobRepository extends Repository<ScheduledJob> {
 	 * materialization skips them and claims different jobs.
 	 * SQLite can't lock rows, but its transactions are `BEGIN IMMEDIATE`, which serializes them to the same effect.
 	 *
+	 * @param lookaheadMs claim a job up to this far before its `nextRunAt`, not only once
+	 * it's already due, so a fixed-interval poll doesn't notice it a whole tick late.
 	 * @returns `undefined` when nothing is due.
 	 *
 	 */
 	async claimDue(
 		manager: EntityManager,
 		limit: number,
+		lookaheadMs = 0,
 	): Promise<{ now: Date; jobs: ScheduledJob[] } | undefined> {
 		const nowExpression = dbNowLiteral(this.isPostgres);
+		const dueExpression = dbNowPlusMsLiteral(this.isPostgres, lookaheadMs);
 
 		const query = manager
 			.createQueryBuilder(ScheduledJob, 'job')
 			.addSelect(nowExpression, 'db_now')
 			.where('job.enabled = :enabled', { enabled: true })
 			.andWhere('job.nextRunAt IS NOT NULL')
-			.andWhere(`job.nextRunAt <= ${nowExpression}`)
+			.andWhere(`job.nextRunAt <= ${dueExpression}`)
 			.orderBy('job.nextRunAt', 'ASC')
 			.limit(limit);
 

packages/@n8n/scheduler/src/core/__tests__/executor-lookahead.test.ts

@@ -5,7 +5,7 @@ import {
 	type ClaimDueTasksBatch,
 	type ExecutorTaskStore,
 } from '../executor';
-import { Loop, executorLookaheadSeconds } from '../lifecycle';
+import { Loop, pollLookaheadSeconds } from '../lifecycle';
 import type { ClaimedTask } from '../types';
 
 /**
@@ -17,7 +17,7 @@ import type { ClaimedTask } from '../types';
  * interval·(1 + 2·jitterRatio) apart: a tick can land jitterRatio·interval early
  * and the next that much late. The scripted `random` below forces exactly that
  * worst case, so a task due in the tail of the gap is only claimable on the
- * earlier tick if the lookahead ({@link executorLookaheadSeconds}) covers the
+ * earlier tick if the lookahead ({@link pollLookaheadSeconds}) covers the
  * whole span. A lookahead that budgets only one side of the jitter leaves the
  * task to the next tick, by which point it is past due and fires late.
  */
@@ -113,7 +113,7 @@ describe('executor claims far enough ahead to fire on time', () => {
 
 		const executor = new Executor(store, registry, new PrecisionTimer(), {
 			leaseSeconds: 60,
-			lookaheadSeconds: executorLookaheadSeconds(INTERVAL_MS / 1_000, JITTER_RATIO),
+			lookaheadSeconds: pollLookaheadSeconds(INTERVAL_MS / 1_000, JITTER_RATIO),
 			batchSize: 100,
 		});
 

packages/@n8n/scheduler/src/core/__tests__/factory.test.ts

@@ -6,9 +6,11 @@ import { SCHEDULER_ATTRIBUTES, SCHEDULER_FIRE_OUTCOME } from '../../observabilit
 import type { SchedulerMetrics } from '../../observability/metrics';
 import { SpanStatus, type Span, type Tracer } from '../../observability/tracer';
 import { InvalidLifecycleOptionsError } from '../errors';
+import { DEFAULT_EXECUTOR_OPTIONS } from '../executor';
 import { createScheduler, DEFAULT_DISPATCH_LAG_WARN_THRESHOLD_SECONDS } from '../factory';
 import type { SchedulerDeps, SchedulerEvent, SchedulerTaskStore } from '../factory';
-import { PASS_TIMED_OUT } from '../lifecycle';
+import { DEFAULT_LIFECYCLE_OPTIONS, PASS_TIMED_OUT, pollLookaheadSeconds } from '../lifecycle';
+import { DEFAULT_MATERIALIZER_OPTIONS } from '../materializer';
 import type { MaterializerTransaction, RunInTransaction } from '../materializer';
 import type { ExpiredLeaseRow } from '../reaper';
 import { DEFAULT_RETENTION_OPTIONS } from '../retention';
@@ -180,6 +182,52 @@ describe('createScheduler executor config', () => {
 	});
 });
 
+// The lookahead createScheduler derives for the materializer: the poll's own
+// worst-case tick gap plus the executor's lookahead (defaults: 10s·1.2 + 5s = 17s).
+const DERIVED_MATERIALIZER_LOOKAHEAD_SECONDS =
+	pollLookaheadSeconds(
+		DEFAULT_LIFECYCLE_OPTIONS.materializerIntervalSeconds,
+		DEFAULT_LIFECYCLE_OPTIONS.jitterRatio,
+	) + DEFAULT_EXECUTOR_OPTIONS.lookaheadSeconds;
+
+const MATERIALIZER_WINDOW_WARNING =
+	'Scheduler materializer lookahead exceeds the window; jobs may be reclaimed with nothing to plan';
+
+describe('createScheduler materializer config', () => {
+	it('emits a warn event at composition when the derived lookahead exceeds the window', () => {
+		// A window shorter than the derived lookahead: materialization degrades to
+		// reclaiming a job every poll with nothing new to plan.
+		const { onEvent } = makeScheduler({ materializer: { windowSeconds: 5 } });
+
+		expect(onEvent).toHaveBeenCalledWith({
+			level: 'warn',
+			message: MATERIALIZER_WINDOW_WARNING,
+			context: { lookaheadSeconds: DERIVED_MATERIALIZER_LOOKAHEAD_SECONDS, windowSeconds: 5 },
+		});
+	});
+
+	it('stays silent at the boundary where the window exactly equals the lookahead', () => {
+		// A job claimed at `now + lookahead == windowEnd` still records its own
+		// occurrence, so equality is not yet degenerate: warn only past the window.
+		const { onEvent } = makeScheduler({
+			materializer: { windowSeconds: DERIVED_MATERIALIZER_LOOKAHEAD_SECONDS },
+		});
+
+		expect(onEvent).not.toHaveBeenCalledWith(
+			expect.objectContaining({ message: MATERIALIZER_WINDOW_WARNING }),
+		);
+	});
+
+	it('stays silent when the window comfortably covers the lookahead', () => {
+		// The default 60s window against the ~17s derived lookahead: no warning.
+		const { onEvent } = makeScheduler();
+
+		expect(onEvent).not.toHaveBeenCalledWith(
+			expect.objectContaining({ message: MATERIALIZER_WINDOW_WARNING }),
+		);
+	});
+});
+
 /** A task claimed for this host whose `runAt` has passed, so it fires on the next tick. */
 const claimedTask = (overrides: Partial<ClaimedTask> = {}): ClaimedTask => ({
 	id: '1',
@@ -354,6 +402,27 @@ describe('createScheduler materialize', () => {
 			context: { planned: 1, recorded: 0 },
 		});
 	});
+
+	it('claims jobs ahead of due by the materializer tick gap plus the executor lookahead', async () => {
+		// Regression guard for window-boundary dispatch lag: the materializer polls on a
+		// fixed, jittered tick, so claiming strictly at `nextRunAt <= now` would notice a
+		// boundary job a whole tick late. createScheduler must derive the claim lookahead,
+		// wide enough that the recorded row also reaches the executor before it fires, not
+		// leave it 0.
+		const tx = mock<MaterializerTransaction>();
+		tx.claimDueJobs.mockResolvedValue(undefined);
+		const materializerTransaction: RunInTransaction = async (work) => await work(tx);
+		const { scheduler } = makeScheduler({ materializerTransaction });
+
+		await scheduler.materialize();
+
+		const expectedLookaheadMs = DERIVED_MATERIALIZER_LOOKAHEAD_SECONDS * 1000;
+		expect(expectedLookaheadMs).toBeGreaterThan(0);
+		expect(tx.claimDueJobs).toHaveBeenCalledWith(
+			DEFAULT_MATERIALIZER_OPTIONS.batchSize,
+			expectedLookaheadMs,
+		);
+	});
 });
 
 describe('createScheduler lifecycle', () => {

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

@@ -15,7 +15,7 @@ import {
 	TaskHandlerRegistry,
 } from './executor';
 import type { ExecutorOptions, ExecutorTaskStore } from './executor';
-import { DEFAULT_LIFECYCLE_OPTIONS, Loop, PASS_TIMED_OUT } from './lifecycle';
+import { DEFAULT_LIFECYCLE_OPTIONS, pollLookaheadSeconds, Loop, PASS_TIMED_OUT } from './lifecycle';
 import type { LifecycleOptions } from './lifecycle';
 import { DEFAULT_MATERIALIZER_OPTIONS, materialize } from './materializer';
 import type { MaterializerOptions, RunInTransaction } from './materializer';
@@ -197,6 +197,34 @@ export function createScheduler(deps: SchedulerDeps): Scheduler & SchedulerPasse
 	};
 	const described = (error: unknown) => ensureError(error).message;
 
+	// Derived, not caller-chosen: the materializer must record a job's occurrences
+	// early enough that the executor still has them in hand when it needs to fire.
+	// That spans two gaps: the materializer loop's own worst-case tick gap (so a job
+	// due between ticks is claimed, not noticed a tick late), plus the executor's
+	// lookahead (so the recorded task row exists before the executor pre-arms its
+	// timer). Drop the executor term and a boundary occurrence can land as late as
+	// its `nextRunAt`, leaving the executor no slack and firing it up to one executor
+	// tick late.
+	materializerOptions.lookaheadSeconds =
+		pollLookaheadSeconds(
+			lifecycleOptions.materializerIntervalSeconds,
+			lifecycleOptions.jitterRatio,
+		) + executorOptions.lookaheadSeconds;
+	if (materializerOptions.lookaheadSeconds > materializerOptions.windowSeconds) {
+		// The lookahead is meant to sit inside the window: claim a job a little before
+		// its window lapses so the next occurrences are planned ahead. A lookahead past
+		// the whole window means a job is re-claimed every poll with nothing new to
+		// record, degrading to no-lookahead materialization.
+		emit(
+			'warn',
+			'Scheduler materializer lookahead exceeds the window; jobs may be reclaimed with nothing to plan',
+			{
+				lookaheadSeconds: materializerOptions.lookaheadSeconds,
+				windowSeconds: materializerOptions.windowSeconds,
+			},
+		);
+	}
+
 	// Metrics are best-effort observability: a throwing sink (e.g. a broken
 	// exporter) must never break the pass that emitted, so every record is
 	// wrapped and its failure swallowed.

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

@@ -51,7 +51,7 @@ export type {
 	NewOccurrence,
 	RunInTransaction,
 } from './materializer';
-export { executorLookaheadSeconds } from './lifecycle';
+export { pollLookaheadSeconds } from './lifecycle';
 export type { ConcurrencyMode, LifecycleOptions } from './lifecycle';
 export type { ReaperOptions, ReapResult } from './reaper';
 export type { RetentionOptions, RetentionSummary } from './retention';

packages/@n8n/scheduler/src/core/lifecycle/__tests__/lookahead.test.ts

@@ -1,28 +1,28 @@
-import { executorLookaheadSeconds } from '../lookahead';
+import { pollLookaheadSeconds } from '../lookahead';
 
-describe('executorLookaheadSeconds', () => {
+describe('pollLookaheadSeconds', () => {
 	const INTERVAL = 10;
 
 	it('is exactly the interval when there is no jitter', () => {
-		expect(executorLookaheadSeconds(INTERVAL, 0)).toBe(10);
+		expect(pollLookaheadSeconds(INTERVAL, 0)).toBe(10);
 	});
 
 	it('adds twice the jitter ratio, covering both the early and late side', () => {
 		// Twice the jitter rather than once: a one-sided budget would give 11 and fire tail tasks late.
-		expect(executorLookaheadSeconds(INTERVAL, 0.1)).toBeCloseTo(12);
-		expect(executorLookaheadSeconds(INTERVAL, 0.25)).toBeCloseTo(15);
-		expect(executorLookaheadSeconds(INTERVAL, 0.5)).toBeCloseTo(20);
+		expect(pollLookaheadSeconds(INTERVAL, 0.1)).toBeCloseTo(12);
+		expect(pollLookaheadSeconds(INTERVAL, 0.25)).toBeCloseTo(15);
+		expect(pollLookaheadSeconds(INTERVAL, 0.5)).toBeCloseTo(20);
 	});
 
 	it('always exceeds a one-sided (jitter-only) budget while jitter is non-zero', () => {
 		for (const jitter of [0.01, 0.1, 0.3, 0.99]) {
 			const oneSided = INTERVAL * (1 + jitter);
-			expect(executorLookaheadSeconds(INTERVAL, jitter)).toBeGreaterThan(oneSided);
+			expect(pollLookaheadSeconds(INTERVAL, jitter)).toBeGreaterThan(oneSided);
 		}
 	});
 
 	it('scales linearly with the interval', () => {
-		expect(executorLookaheadSeconds(20, 0.1)).toBeCloseTo(24);
-		expect(executorLookaheadSeconds(5, 0.1)).toBeCloseTo(6);
+		expect(pollLookaheadSeconds(20, 0.1)).toBeCloseTo(24);
+		expect(pollLookaheadSeconds(5, 0.1)).toBeCloseTo(6);
 	});
 });

packages/@n8n/scheduler/src/core/lifecycle/__tests__/timeline.test.ts

@@ -1,4 +1,4 @@
-import { executorLookaheadSeconds } from '../lookahead';
+import { pollLookaheadSeconds } from '../lookahead';
 import { Timeline } from '../timeline';
 
 const INTERVAL = 10_000;
@@ -56,14 +56,14 @@ describe('Timeline', () => {
 
 			expect(maxGap).toBe(INTERVAL * (1 + 2 * JITTER)); // 12_000
 			// The lookahead must cover the widest gap, and does so exactly (tight, not padded).
-			expect(executorLookaheadSeconds(INTERVAL / 1_000, JITTER) * 1_000).toBe(maxGap);
+			expect(pollLookaheadSeconds(INTERVAL / 1_000, JITTER) * 1_000).toBe(maxGap);
 		});
 
 		it('covers every gap under arbitrary jitter, coupling the timeline and the lookahead', () => {
 			// A spread of jitter draws, including the extremes that reach the bound. If either
 			// the jitter span or the lookahead formula drifts out of step, a gap escapes here.
 			const draws = [0, 1, 0.5, 0.9, 0.1, 1, 0, 0.3, 0.7, 0, 1, 0.5];
-			const lookaheadMs = executorLookaheadSeconds(INTERVAL / 1_000, JITTER) * 1_000;
+			const lookaheadMs = pollLookaheadSeconds(INTERVAL / 1_000, JITTER) * 1_000;
 
 			for (const gap of gaps(fires(timelineWith([0.5, ...draws]), draws.length + 1))) {
 				expect(gap).toBeLessThanOrEqual(lookaheadMs);

packages/@n8n/scheduler/src/core/lifecycle/index.ts

@@ -1,5 +1,5 @@
 export { Loop, PASS_TIMED_OUT } from './loop';
 export type { ConcurrencyMode, LoopHooks, LoopOptions } from './loop';
-export { executorLookaheadSeconds } from './lookahead';
+export { pollLookaheadSeconds } from './lookahead';
 export { DEFAULT_LIFECYCLE_OPTIONS } from './options';
 export type { LifecycleOptions } from './options';

packages/@n8n/scheduler/src/core/lifecycle/lookahead.ts

@@ -1,14 +1,14 @@
 /**
- * How far ahead of a tick the executor must claim so a task due before the next
- * tick is claimed in time to fire precisely on the timeline (see {@link Timeline}).
+ * How far ahead of a tick a fixed-interval poll loop must claim so a row due
+ * before its next tick is claimed in time (used by the executor to fire on the
+ * timeline, and by the materializer to plan a job before its window runs out).
  *
- * The horizon has to cover the largest gap between two consecutive fires. Jitter
+ * The horizon has to cover the largest gap between two consecutive ticks. Jitter
  * is applied per slot and symmetric, so one tick can land up to
  * `jitterRatio·interval` early and the next up to `jitterRatio·interval` late:
  * the gap stretches by both sides at once, to `interval·(1 + 2·jitterRatio)`.
- * Budgeting only one side leaves a task due in that tail claimed a tick late,
- * so it fires late.
+ * Budgeting only one side leaves a row due in that tail claimed a tick late.
  */
-export function executorLookaheadSeconds(intervalSeconds: number, jitterRatio: number): number {
+export function pollLookaheadSeconds(intervalSeconds: number, jitterRatio: number): number {
 	return intervalSeconds * (1 + 2 * jitterRatio);
 }

packages/@n8n/scheduler/src/core/materializer/__tests__/materialize.test.ts

@@ -31,6 +31,7 @@ const runnerWith =
 
 const options: MaterializerOptions = {
 	windowSeconds: 0,
+	lookaheadSeconds: 0,
 	batchSize: 25,
 	maxPerJob: 100,
 	planRetrySeconds: 3600,
@@ -92,7 +93,18 @@ describe('materialize', () => {
 
 		await materialize(runnerWith(tx), { ...options, batchSize: 25 });
 
-		expect(tx.claimDueJobs).toHaveBeenCalledWith(25);
+		expect(tx.claimDueJobs).toHaveBeenCalledWith(25, 0);
+	});
+
+	it('claims ahead by lookaheadSeconds, passing it to the claim in milliseconds', async () => {
+		const tx = mock<MaterializerTransaction>();
+		tx.claimDueJobs.mockResolvedValue(undefined);
+
+		await materialize(runnerWith(tx), { ...options, batchSize: 25, lookaheadSeconds: 12 });
+
+		// 12s of lookahead reaches the claim as 12_000ms, so a job due within the next
+		// poll interval is claimed now instead of a whole tick after it comes due.
+		expect(tx.claimDueJobs).toHaveBeenCalledWith(25, 12_000);
 	});
 
 	it('reports skipped duplicates, and a throwing reporter does not fail the pass', async () => {

packages/@n8n/scheduler/src/core/materializer/materialize.ts

@@ -91,7 +91,8 @@ export async function materialize(
 ): Promise<MaterializerSummary> {
 	signal?.throwIfAborted();
 	return await runInTransaction<MaterializerSummary>(async (tx) => {
-		const claimed = await tx.claimDueJobs(options.batchSize);
+		const lookaheadMs = options.lookaheadSeconds * Time.seconds.toMilliseconds;
+		const claimed = await tx.claimDueJobs(options.batchSize, lookaheadMs);
 		signal?.throwIfAborted();
 		if (claimed === undefined) {
 			return { claimedJobs: 0, occurrences: 0, created: [], deferredJobs: 0 };

packages/@n8n/scheduler/src/core/materializer/options.ts

@@ -11,6 +11,17 @@ export interface MaterializerOptions {
 	 */
 	windowSeconds: number;
 
+	/**
+	 * How far before a job's `nextRunAt` the claim query already picks it up, in
+	 * seconds. Without it, `claimDue` only claims once `nextRunAt <= now`, so a
+	 * job already due waits for the next materializer poll tick to be noticed:
+	 * up to one poll interval of dispatch lag on top of the job's own schedule.
+	 * The host derives it, not a caller-chosen constant: the materializer poll's
+	 * own worst-case tick gap plus the executor's lookahead, so a job's
+	 * occurrences are recorded early enough for the executor to fire them on time.
+	 */
+	lookaheadSeconds: number;
+
 	/**
 	 * The most occurrences to record for one job in one pass
 	 * (drains a backlog in batches).
@@ -37,6 +48,7 @@ export interface MaterializerOptions {
 
 export const DEFAULT_MATERIALIZER_OPTIONS: MaterializerOptions = {
 	windowSeconds: 60,
+	lookaheadSeconds: 0,
 	batchSize: 100,
 	maxPerJob: 1000,
 	planRetrySeconds: Time.hours.toSeconds,

packages/@n8n/scheduler/src/core/materializer/transaction.ts

@@ -50,11 +50,14 @@ export interface DueJobs {
  */
 export interface MaterializerTransaction {
 	/**
-	 * @returns up to `limit` enabled jobs whose next run is due, oldest first, locking
-	 * them so a concurrent pass claims different jobs, with the database time they were
-	 * judged due at; `undefined` when nothing is due.
+	 * @param lookaheadMs claim a job up to this far before its `nextRunAt`, not only
+	 * once it's already due, so a fixed-interval poll doesn't add a full tick of its
+	 * own on top of the job's schedule (see `MaterializerOptions.lookaheadSeconds`).
+	 * @returns up to `limit` enabled jobs whose next run is due (within `lookaheadMs`),
+	 * oldest first, locking them so a concurrent pass claims different jobs, with the
+	 * database time they were judged due at; `undefined` when nothing is due.
 	 */
-	claimDueJobs(limit: number): Promise<DueJobs | undefined>;
+	claimDueJobs(limit: number, lookaheadMs: number): Promise<DueJobs | undefined>;
 
 	/**
 	 * Record every planned occurrence across all jobs in one batch,

packages/cli/src/scheduling/__tests__/durable-scheduler.test.ts

@@ -12,7 +12,7 @@ import { DurableScheduler } from '../durable-scheduler';
 import { SCHEDULE_TRIGGER_TASK_TYPE } from '../schedule-trigger-node/schedule-trigger-task';
 import type { ScheduleTriggerTaskHandler } from '../schedule-trigger-node/schedule-trigger-task-handler';
 
-// Keep the real exports (e.g. executorLookaheadSeconds) so the wiring is tested
+// Keep the real exports (e.g. pollLookaheadSeconds) so the wiring is tested
 // against the actual formula; only the scheduler factory is stubbed.
 vi.mock('@n8n/scheduler', async (importOriginal) => ({
 	...(await importOriginal<typeof import('@n8n/scheduler')>()),

packages/cli/src/scheduling/durable-scheduler.ts

@@ -4,7 +4,7 @@ import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8
 import { OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import type { RunInTransaction, Scheduler, TaskHandler } from '@n8n/scheduler';
-import { createScheduler, executorLookaheadSeconds } from '@n8n/scheduler';
+import { createScheduler, pollLookaheadSeconds } from '@n8n/scheduler';
 import { InstanceSettings, Tracing } from 'n8n-core';
 
 import { PrometheusSchedulerMetricsService } from '@/metrics/prometheus/scheduler-metrics.service';
@@ -52,7 +52,7 @@ export class DurableScheduler implements Scheduler {
 						leaseSeconds: config.leaseDurationSeconds,
 						// Claim one executor tick ahead so a task due before the next tick still
 						// fires on time; the horizon must cover the widest gap between two ticks.
-						lookaheadSeconds: executorLookaheadSeconds(
+						lookaheadSeconds: pollLookaheadSeconds(
 							config.executorIntervalSeconds,
 							config.jitterRatio,
 						),
@@ -117,7 +117,8 @@ export function buildMaterializerTransaction(
 		await dataSource.transaction(
 			async (manager) =>
 				await work({
-					claimDueJobs: async (limit) => await jobs.claimDue(manager, limit),
+					claimDueJobs: async (limit, lookaheadMs) =>
+						await jobs.claimDue(manager, limit, lookaheadMs),
 					recordOccurrences: async (occurrences) =>
 						await tasks.insertIgnoringDuplicates(manager, occurrences),
 					advanceJobs: async (planned) => {

packages/cli/test/integration/scheduling/scheduled-repositories.test.ts

@@ -197,6 +197,32 @@ describe('scheduled repositories', () => {
 			expect(claimed?.jobs.map((j) => j.id)).toEqual([dueEarly.id, dueLate.id]);
 		});
 
+		it('claims a not-yet-due job whose nextRunAt falls within the lookahead', async () => {
+			// The materializer polls on a fixed tick, so a job due just after a tick would
+			// otherwise wait a whole interval to be noticed. Claiming ahead of due lets the
+			// job be planned before its window lapses. Default (0) lookahead ignores it.
+			const soon = await createJob({ nextRunAt: secondsFromNow(5) });
+
+			const strict = await dataSource.transaction(
+				async (trx) => await jobRepository.claimDue(trx, 100),
+			);
+			expect(strict).toBeUndefined();
+
+			const withLookahead = await dataSource.transaction(
+				async (trx) => await jobRepository.claimDue(trx, 100, 10_000),
+			);
+			expect(withLookahead?.jobs.map((j) => j.id)).toEqual([soon.id]);
+		});
+
+		it('still excludes a job whose nextRunAt is beyond the lookahead', async () => {
+			await createJob({ nextRunAt: secondsFromNow(60) });
+
+			const claimed = await dataSource.transaction(
+				async (trx) => await jobRepository.claimDue(trx, 100, 10_000),
+			);
+			expect(claimed).toBeUndefined();
+		});
+
 		it('excludes disabled, future, and null-nextRunAt jobs', async () => {
 			await createJob({ enabled: false, nextRunAt: secondsFromNow(-60) });
 			await createJob({ nextRunAt: secondsFromNow(3600) });

packages/cli/test/integration/scheduling/scheduler-materialize.test.ts

@@ -107,16 +107,19 @@ describe('scheduler materialization', () => {
 		expect(first.occurrences).toBe(5);
 		expect(await taskRepo.count()).toBe(5);
 
-		// Successive passes continue draining, each recording at most maxPerJob, until nothing is due.
+		// Successive passes continue draining, each recording at most maxPerJob, until the
+		// backlog is exhausted. The claim reaches a poll interval ahead of due, so with
+		// windowSeconds: 0 the job keeps being claimed for its near-future fire even once
+		// drained; exhaustion shows as a pass that records nothing new, not an unclaimed job.
 		for (let i = 0; i < 10; i++) {
 			const summary = await drainScheduler.materialize();
 			expect(summary.occurrences).toBeLessThanOrEqual(5);
-			if (summary.claimedJobs === 0) break;
+			if (summary.occurrences === 0) break;
 		}
 
 		// Drained: the backlog is fully recorded, every occurrence distinct (no duplicate from batching).
 		const drained = await drainScheduler.materialize();
-		expect(drained.claimedJobs).toBe(0);
+		expect(drained.occurrences).toBe(0);
 		const tasks = await taskRepo.find();
 		const distinctInstants = new Set(tasks.map((t) => t.scheduledFor.getTime()));
 		expect(distinctInstants.size).toBe(tasks.length);