feat(core): Stop a scheduled task run at its timeout (no-changelog) (#40515)

Co-authored-by: Veljko Terzic <veljko.terzic@n8n.io>

packages/@n8n/scheduler/README.md

@@ -411,8 +411,11 @@ The design leans on a few ideas working together.
   matches, the executor counts the failed attempt. It retries only while
   attempts remain. If the claim no longer matches, the write changes nothing.
   A clean return completes an occurrence that the executor still owns.
-  A run still running after sixty leases, or after about 24 days if that comes
-  first, logs a warning, since it may be stuck.
+  Each job stores a timeout, and each occurrence copies it. A run that reaches
+  it gets its signal aborted with a `TaskTimeoutError`, and its lease is no
+  longer renewed, so the recovery pass takes back a handler that ignores the
+  signal. A run that did not dispatch counts a failed attempt. A run that
+  dispatched completes.
 - **Fencing.** Each claim carries a version number (an *epoch*) that increases every
   time a run is claimed. Every final write ("mark succeeded", "mark failed") is
   guarded by that number. So if a slow server comes back from the dead after its

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

@@ -1800,7 +1800,7 @@ describe('createScheduler metrics', () => {
 		const RENEWAL_INTERVAL_MS = 5_000;
 
 		/** Fires one long handler that stays pending until `finish` is called. */
-		const fireLongHandler = async (metrics: SchedulerMetrics) => {
+		const fireLongHandler = async (metrics: SchedulerMetrics, task = claimedTask()) => {
 			const made = makeScheduler({
 				metrics,
 				executor: { leaseSeconds: LEASE_SECONDS, lookaheadSeconds: 1 },
@@ -1814,7 +1814,7 @@ describe('createScheduler metrics', () => {
 					return report.notDispatched();
 				},
 			});
-			made.taskStore.claimDueTasks.mockResolvedValue([claimedTask()]);
+			made.taskStore.claimDueTasks.mockResolvedValue([task]);
 			made.taskStore.beginDispatch.mockResolvedValue(1);
 			made.taskStore.completeTask.mockResolvedValue(1);
 			await made.scheduler.execute();
@@ -1873,27 +1873,27 @@ describe('createScheduler metrics', () => {
 			finish();
 		});
 
-		it('warns once when a run is still pending after sixty leases', async () => {
-			const { taskStore, onEvent, finish } = await fireLongHandler(mock<SchedulerMetrics>());
+		it('counts and warns once when a run reaches its timeout', async () => {
+			const metrics = mock<SchedulerMetrics>();
+			const { taskStore, onEvent, finish } = await fireLongHandler(metrics);
 			taskStore.renewLease.mockResolvedValue(true);
-			const stuckWarning = {
+			const { timeoutSeconds } = claimedTask();
+			const timeoutWarning = {
 				level: 'warn',
-				message: 'Scheduler task is still running after many leases; it may be stuck',
-				context: {
-					taskId: claimedTask().id,
-					taskType: 'test-task',
-					runningSeconds: 60 * LEASE_SECONDS,
-				},
+				message: 'Scheduler task run reached its timeout',
+				context: { taskId: claimedTask().id, taskType: 'test-task', timeoutSeconds },
 			};
 
-			await vi.advanceTimersByTimeAsync(60 * LEASE_SECONDS * 1_000 - 1);
-			expect(onEvent).not.toHaveBeenCalledWith(stuckWarning);
+			await vi.advanceTimersByTimeAsync(timeoutSeconds * 1_000 - 1);
+			expect(onEvent).not.toHaveBeenCalledWith(timeoutWarning);
+			expect(metrics.recordTaskTimeout).not.toHaveBeenCalled();
 
 			await vi.advanceTimersByTimeAsync(60_000);
 			const warnings = onEvent.mock.calls.filter(
-				([event]) => event.message === stuckWarning.message,
+				([event]) => event.message === timeoutWarning.message,
 			);
-			expect(warnings).toEqual([[stuckWarning]]);
+			expect(warnings).toEqual([[timeoutWarning]]);
+			expect(metrics.recordTaskTimeout).toHaveBeenCalledExactlyOnceWith('test-task');
 			finish();
 		});
 	});

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

@@ -83,6 +83,17 @@ export class InvalidRunOptionError extends Error {
 	}
 }
 
+/**
+ * The abort reason of a handler's signal when its run reaches the timeout of
+ * its occurrence. The executor stops renewing the lease at the same time.
+ */
+export class TaskTimeoutError extends Error {
+	constructor(readonly timeoutSeconds: number) {
+		super('The run of this occurrence reached its timeout');
+		this.name = 'TaskTimeoutError';
+	}
+}
+
 /** Raised when a lease heartbeat is given a lease duration that is not a positive integer. */
 export class InvalidLeaseDurationError extends Error {
 	constructor(readonly leaseDurationMs: number) {

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

@@ -1,6 +1,6 @@
 import { mock } from 'vitest-mock-extended';
 
-import { LeaseLostError } from '../../errors';
+import { LeaseLostError, TaskTimeoutError } from '../../errors';
 import type { ClaimedTask } from '../../types';
 import { backoff } from '../backoff';
 import { Executor } from '../executor';
@@ -8,7 +8,7 @@ import type { ExecutorHooks } from '../executor';
 import type { ExecutorOptions } from '../options';
 import type { PrecisionTimer } from '../precision-timer';
 import type { ExecutorTaskStore } from '../store';
-import type { DispatchReporter, TaskHandler, TaskHandlerRegistry } from '../task-handler';
+import type { DispatchReporter, TaskHandler, TaskHandlerRegistry, TaskRun } from '../task-handler';
 import type { ExecutorTracing, FireResult } from '../tracing';
 
 const HOST = 'main-abc';
@@ -54,7 +54,7 @@ const setup = (options?: Partial<ExecutorOptions>) => {
 		onLeaseLost: vi.fn(),
 		onLeaseRenewal: vi.fn(),
 		onLeaseRenewalError: vi.fn(),
-		onLongRunningTask: vi.fn(),
+		onTaskTimeout: vi.fn(),
 	} satisfies ExecutorHooks;
 	// A see-through tracing hook that records its calls and just runs the fire.
 	// The executor's only obligation is to route every fire through this hook;
@@ -250,7 +250,11 @@ describe('Executor.fire', () => {
 		const result = await executor.fire(HOST, task);
 
 		expect(result).toEqual({ outcome: 'completed' });
-		expect(handler.execute).toHaveBeenCalledWith(task, expect.any(Object), expect.any(AbortSignal));
+		expect(handler.execute).toHaveBeenCalledWith(
+			task,
+			expect.any(Object),
+			expect.objectContaining({ signal: expect.any(AbortSignal) }),
+		);
 		expect(store.completeTask).toHaveBeenCalledWith({
 			host: HOST,
 			id: task.id,
@@ -805,20 +809,22 @@ describe('Executor.fire lease renewal', () => {
 		let finish!: () => void;
 		let signal!: AbortSignal;
 		const handler: TaskHandler = {
-			execute: vi.fn(async (_task: ClaimedTask, report: DispatchReporter, s: AbortSignal) => {
-				signal = s;
-				if (dispatchFirst) {
-					report.dispatched();
-				}
-				await new Promise<void>((resolve) => {
-					finish = resolve;
-					s.addEventListener('abort', () => resolve());
-				});
-				if (s.aborted) {
-					throw s.reason;
-				}
-				return report.notDispatched();
-			}),
+			execute: vi.fn(
+				async (_task: ClaimedTask, report: DispatchReporter, { signal: s }: TaskRun) => {
+					signal = s;
+					if (dispatchFirst) {
+						report.dispatched();
+					}
+					await new Promise<void>((resolve) => {
+						finish = resolve;
+						s.addEventListener('abort', () => resolve());
+					});
+					if (s.aborted) {
+						throw s.reason;
+					}
+					return report.notDispatched();
+				},
+			),
 		};
 		return { handler, finish: () => finish(), signal: () => signal };
 	};
@@ -994,52 +1000,146 @@ describe('Executor.fire lease renewal', () => {
 		},
 	);
 
-	it('reports a run still pending after sixty leases, and not one that settled earlier', async () => {
-		const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
-		const stuck = longRunning();
-		const quick = longRunning();
-		registry.resolve.mockReturnValueOnce(quick.handler).mockReturnValueOnce(stuck.handler);
-		store.beginDispatch.mockResolvedValue(1);
-		store.renewLease.mockResolvedValue(true);
-		store.completeTask.mockResolvedValue(1);
-		const quickTask = claimedTask({ id: 'quick' });
-		const stuckTask = claimedTask({ id: 'stuck' });
+	describe('timeout', () => {
+		const TIMEOUT_SECONDS = 4 * LEASE_SECONDS;
+		const TIMEOUT_MS = TIMEOUT_SECONDS * 1_000;
 
-		const quickFiring = executor.fire(HOST, quickTask);
-		await vi.advanceTimersByTimeAsync(RENEWAL_INTERVAL_MS);
-		quick.finish();
-		await quickFiring;
-		const stuckFiring = executor.fire(HOST, stuckTask);
-		await vi.advanceTimersByTimeAsync(60 * LEASE_SECONDS * 1_000 - 1);
-		expect(hooks.onLongRunningTask).not.toHaveBeenCalled();
+		it('aborts a run that did not dispatch with a TaskTimeoutError, stops renewing, and counts a failed attempt', async () => {
+			const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			const { handler, signal } = longRunning();
+			registry.resolve.mockReturnValue(handler);
+			store.beginDispatch.mockResolvedValue(1);
+			store.renewLease.mockResolvedValue(true);
+			store.rescheduleTask.mockResolvedValue(1);
+			const task = claimedTask({ maxAttempts: 3, timeoutSeconds: TIMEOUT_SECONDS });
 
-		await vi.advanceTimersByTimeAsync(1);
-		expect(hooks.onLongRunningTask).toHaveBeenCalledExactlyOnceWith(stuckTask, 60 * LEASE_SECONDS);
+			const firing = executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS - 1);
+			expect(signal().aborted).toBe(false);
+			expect(hooks.onTaskTimeout).not.toHaveBeenCalled();
 
-		stuck.finish();
-		await stuckFiring;
-	});
+			await vi.advanceTimersByTimeAsync(1);
+			expect(signal().reason).toBeInstanceOf(TaskTimeoutError);
+			expect(signal().reason).toMatchObject({ timeoutSeconds: TIMEOUT_SECONDS });
+			expect(hooks.onTaskTimeout).toHaveBeenCalledExactlyOnceWith(task);
+			await expect(firing).resolves.toMatchObject({ outcome: 'rescheduled' });
+			expect(store.rescheduleTask).toHaveBeenCalledExactlyOnceWith(
+				{ host: HOST, id: task.id, claimedEpoch: task.leaseEpoch },
+				backoff(1),
+				'The run of this occurrence reached its timeout',
+			);
+			const renewals = store.renewLease.mock.calls.length;
+			await vi.advanceTimersByTimeAsync(2 * LEASE_SECONDS * 1_000);
+			expect(store.renewLease).toHaveBeenCalledTimes(renewals);
+		});
 
-	it('reports the time actually waited when sixty leases exceed the timer limit', async () => {
-		const leaseSeconds = 24 * 60 * 60;
-		const { store, registry, hooks, executor } = setup({ leaseSeconds });
-		const { handler, finish } = longRunning();
-		registry.resolve.mockReturnValue(handler);
-		store.beginDispatch.mockResolvedValue(1);
-		store.renewLease.mockResolvedValue(true);
-		store.completeTask.mockResolvedValue(1);
-		const task = claimedTask();
-		const maxTimeoutMs = 2 ** 31 - 1;
+		it('hands the handler the time left before its signal aborts, counted from before the dispatch write', async () => {
+			const { store, registry, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			const { handler, signal } = longRunning();
+			registry.resolve.mockReturnValue(handler);
+			const DISPATCH_DELAY_MS = 2_000;
+			store.beginDispatch.mockImplementation(
+				async () => await new Promise((resolve) => setTimeout(() => resolve(1), DISPATCH_DELAY_MS)),
+			);
+			store.renewLease.mockResolvedValue(true);
+			store.rescheduleTask.mockResolvedValue(1);
+			const task = claimedTask({ maxAttempts: 3, timeoutSeconds: TIMEOUT_SECONDS });
 
-		const firing = executor.fire(HOST, task);
-		await vi.advanceTimersByTimeAsync(1_000);
-		expect(hooks.onLongRunningTask).not.toHaveBeenCalled();
+			const firing = executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(DISPATCH_DELAY_MS);
+			const run = vi.mocked(handler.execute).mock.calls[0][2];
+			expect(run.remainingMs()).toBe(TIMEOUT_MS - DISPATCH_DELAY_MS);
 
-		await vi.advanceTimersByTimeAsync(maxTimeoutMs - 1_000);
-		expect(hooks.onLongRunningTask).toHaveBeenCalledExactlyOnceWith(task, maxTimeoutMs / 1_000);
+			await vi.advanceTimersByTimeAsync(run.remainingMs() - 1);
+			expect(signal().aborted).toBe(false);
+			await vi.advanceTimersByTimeAsync(1);
+			expect(signal().reason).toBeInstanceOf(TaskTimeoutError);
+			expect(run.remainingMs()).toBe(0);
+			await firing;
+		});
 
-		finish();
-		await firing;
+		it('stops renewing a handler that ignores its signal, so its lease expires', async () => {
+			const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			let signal!: AbortSignal;
+			registry.resolve.mockReturnValue({
+				execute: vi.fn(async (_task: ClaimedTask, _report: DispatchReporter, run: TaskRun) => {
+					signal = run.signal;
+					return await new Promise<never>(() => {});
+				}),
+			});
+			store.beginDispatch.mockResolvedValue(1);
+			store.renewLease.mockResolvedValue(true);
+			const task = claimedTask({ timeoutSeconds: TIMEOUT_SECONDS });
+
+			void executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS);
+			const renewals = store.renewLease.mock.calls.length;
+			expect(renewals).toBeGreaterThan(0);
+			expect(signal.reason).toBeInstanceOf(TaskTimeoutError);
+
+			await vi.advanceTimersByTimeAsync(2 * LEASE_SECONDS * 1_000);
+			expect(store.renewLease).toHaveBeenCalledTimes(renewals);
+			expect(hooks.onLeaseRenewal).not.toHaveBeenCalledWith(task, 'expired');
+			expect(store.rescheduleTask).not.toHaveBeenCalled();
+			expect(store.completeTask).not.toHaveBeenCalled();
+		});
+
+		it('completes a dispatched run that reaches its timeout instead of retrying it', async () => {
+			const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			const { handler, signal } = longRunning({ dispatchFirst: true });
+			registry.resolve.mockReturnValue(handler);
+			store.beginDispatch.mockResolvedValue(1);
+			store.renewLease.mockResolvedValue(true);
+			store.completeTask.mockResolvedValue(1);
+			const task = claimedTask({ maxAttempts: 3, timeoutSeconds: TIMEOUT_SECONDS });
+
+			const firing = executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS);
+
+			expect(signal().reason).toBeInstanceOf(TaskTimeoutError);
+			await expect(firing).resolves.toEqual({ outcome: 'completed' });
+			expect(store.rescheduleTask).not.toHaveBeenCalled();
+			expect(hooks.onFire).toHaveBeenCalledWith(task.taskType, 'success');
+		});
+
+		it('leaves a run that settles before its timeout alone', async () => {
+			const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			const { handler, finish, signal } = longRunning();
+			registry.resolve.mockReturnValue(handler);
+			store.beginDispatch.mockResolvedValue(1);
+			store.renewLease.mockResolvedValue(true);
+			store.completeTask.mockResolvedValue(1);
+			const task = claimedTask({ timeoutSeconds: TIMEOUT_SECONDS });
+
+			const firing = executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS - 1);
+			finish();
+			await expect(firing).resolves.toEqual({ outcome: 'completed' });
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS);
+
+			expect(signal().aborted).toBe(false);
+			expect(hooks.onTaskTimeout).not.toHaveBeenCalled();
+		});
+
+		it('does not report a timeout for a run already aborted by a lost claim', async () => {
+			const { store, registry, hooks, executor } = setup({ leaseSeconds: LEASE_SECONDS });
+			let signal!: AbortSignal;
+			registry.resolve.mockReturnValue({
+				execute: vi.fn(async (_task: ClaimedTask, _report: DispatchReporter, run: TaskRun) => {
+					signal = run.signal;
+					return await new Promise<never>(() => {});
+				}),
+			});
+			store.beginDispatch.mockResolvedValue(1);
+			store.renewLease.mockResolvedValue(false);
+			const task = claimedTask({ timeoutSeconds: TIMEOUT_SECONDS });
+
+			void executor.fire(HOST, task);
+			await vi.advanceTimersByTimeAsync(TIMEOUT_MS);
+
+			expect(signal.reason).toBeInstanceOf(LeaseLostError);
+			expect(hooks.onTaskTimeout).not.toHaveBeenCalled();
+		});
 	});
 
 	it('reports a failed renewal without aborting the handler', async () => {

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

@@ -1,9 +1,9 @@
-import { MAX_INTEGER_32BITS_SIGNED, Time } from '@n8n/constants';
+import { Time } from '@n8n/constants';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 
 import { backoff } from './backoff';
-import { LeaseLostError } from '../errors';
-import { LONG_RUN_THRESHOLD_IN_LEASES, MIN_RENEWAL_INTERVAL_MS } from './lease-constants';
+import { LeaseLostError, TaskTimeoutError } from '../errors';
+import { MIN_RENEWAL_INTERVAL_MS } from './lease-constants';
 import { LeaseHeartbeat } from './lease-heartbeat';
 import type { LeaseRenewalResult } from './lease-heartbeat';
 import { DEFAULT_EXECUTOR_OPTIONS, type ExecutorOptions } from './options';
@@ -13,6 +13,7 @@ import { createDispatchReporter } from './task-handler';
 import type { DispatchReporter, TaskHandler, TaskHandlerRegistry } from './task-handler';
 import { noopExecutorTracing } from './tracing';
 import type { ExecutorTracing, FireResult } from './tracing';
+import { Alarm } from '../lifecycle/alarm';
 import type { ClaimedTask } from '../types';
 
 type ClaimedEntry = { host: string; task: ClaimedTask };
@@ -64,8 +65,11 @@ export interface ExecutorHooks {
 	/** A lease renewal write failed; the next renewal tries again. */
 	onLeaseRenewalError?: (task: ClaimedTask, error: unknown) => void;
 
-	/** A handler is still running after many leases, so it may be stuck. Fires once a run. */
-	onLongRunningTask?: (task: ClaimedTask, runningSeconds: number) => void;
+	/**
+	 * A run reached the timeout of its occurrence. Its signal is aborted and its
+	 * lease is no longer renewed. Fires once a run.
+	 */
+	onTaskTimeout?: (task: ClaimedTask) => void;
 
 	// Fire-path metrics hooks (the normal path), distinct from the incident hooks above.
 
@@ -98,7 +102,7 @@ export interface ExecutorHooks {
  * an atomic compare-and-set that stamps `startedAt` and returns 1 for a single
  * winner, so the handler runs at most once per lease. That same write refreshes the
  * lease, and a heartbeat renews it while the handler runs, so a long handler keeps
- * its claim for as long as its instance is alive.
+ * its claim for as long as its instance is alive, up to the timeout of its occurrence.
  *
  * The contract is at-least-once. Ownership only lasts as long as the lease: if an
  * owner is lost past it (crash or partition), the reaper reclaims the row, clears
@@ -311,19 +315,23 @@ export class Executor {
 			);
 		};
 		const report = createDispatchReporter(markDispatched);
+		const run = new AbortController();
 
 		// Record success only after the try, so a failure to record it isn't taken for a
 		// handler failure. Such a failure propagates out (caught by the detached `.catch`
 		// in claimAndSchedule) and leaves the row `running` for the reaper.
 		try {
-			await this.executeWithHeartbeat(handler, task, report, {
+			await this.executeWithHeartbeat(handler, task, report, run, {
 				claim,
 				leaseSetAt,
 				isMarkedDispatched: () => isMarkedDispatched,
 			});
 		} catch (error) {
 			await dispatchMark;
-			return await this.recordHandlerFailure(task, claim, error, dispatchMark !== undefined);
+			return await this.recordHandlerFailure(task, claim, error, {
+				dispatchWasReported: dispatchMark !== undefined,
+				abortReason: run.signal.reason,
+			});
 		}
 
 		markDispatched(); // A handler that returned without throwing is considered as dispatched
@@ -341,11 +349,14 @@ export class Executor {
 		task: ClaimedTask,
 		claim: ClaimedTaskRef,
 		error: unknown,
-		dispatchWasReported: boolean,
+		{ dispatchWasReported, abortReason }: { dispatchWasReported: boolean; abortReason: unknown },
 	): Promise<FireResult> {
 		const errorMessage = ensureError(error).message;
+		const timedOut = abortReason instanceof TaskTimeoutError;
 		const nextAttempts = task.attempts + 1;
-		if (nextAttempts < task.maxAttempts) {
+		// A dispatched run that timed out is completed, as the reaper completes it
+		// when the handler ignores its signal.
+		if (nextAttempts < task.maxAttempts && !(dispatchWasReported && timedOut)) {
 			const rowsAffected = await this.store.rescheduleTask(
 				claim,
 				backoff(nextAttempts),
@@ -359,7 +370,7 @@ export class Executor {
 			return { outcome: 'skipped-not-owned', errorMessage };
 		}
 
-		// On the last attempt, complete work whose effect was handed off before the error.
+		// On the last attempt or a timeout, complete work whose effect was handed off before the error.
 		// The report counts even if its marker write failed.
 		if (dispatchWasReported) {
 			const rowsAffected = await this.store.completeTask(claim);
@@ -383,15 +394,16 @@ export class Executor {
 
 	/**
 	 * Run the handler while a heartbeat renews the claim's lease, until it settles.
-	 * A lost claim aborts the handler's signal unless the dispatch marker is stored.
+	 * A lost claim aborts `run` unless the dispatch marker is stored. The timeout
+	 * of the occurrence aborts `run` and stops the renewals.
 	 */
 	private async executeWithHeartbeat(
 		handler: TaskHandler,
 		task: ClaimedTask,
 		report: DispatchReporter,
+		run: AbortController,
 		{ claim, leaseSetAt, isMarkedDispatched }: HeartbeatRun,
 	): Promise<void> {
-		const lease = new AbortController();
 		const heartbeat = new LeaseHeartbeat(
 			async (expiresInMs) => await this.store.renewLease(claim, expiresInMs),
 			{ leaseDurationMs: this.leaseMs, leaseSetAt },
@@ -403,26 +415,31 @@ export class Executor {
 					// marker write still in flight counts as not stored: the reaper may
 					// already redeliver the row, and that write may never finish.
 					if (result !== 'renewed' && !isMarkedDispatched()) {
-						lease.abort(new LeaseLostError());
+						run.abort(new LeaseLostError());
 					}
 				},
 				onRenewalError: (error) => this.hooks.onLeaseRenewalError?.(task, error),
 			},
 		);
-		// `setTimeout` fires a longer delay at once.
-		const longRunMs = Math.min(
-			LONG_RUN_THRESHOLD_IN_LEASES * this.leaseMs,
-			MAX_INTEGER_32BITS_SIGNED,
-		);
-		const longRunTimer = setTimeout(
-			() => this.hooks.onLongRunningTask?.(task, longRunMs / Time.seconds.toMilliseconds),
-			longRunMs,
-		);
-		longRunTimer.unref();
+		const timeoutMs = task.timeoutSeconds * Time.seconds.toMilliseconds;
+		const deadline = leaseSetAt + timeoutMs;
+		const timeout = new Alarm(() => performance.now());
+		timeout.set(deadline, () => {
+			// Without renewals the lease expires, so the reaper recovers a handler that
+			// ignores its signal.
+			heartbeat.stop();
+			if (!run.signal.aborted) {
+				this.hooks.onTaskTimeout?.(task);
+				run.abort(new TaskTimeoutError(task.timeoutSeconds));
+			}
+		});
 		try {
-			await handler.execute(task, report, lease.signal);
+			await handler.execute(task, report, {
+				signal: run.signal,
+				remainingMs: () => Math.max(0, deadline - performance.now()),
+			});
 		} finally {
-			clearTimeout(longRunTimer);
+			timeout.cancel();
 			heartbeat.stop();
 		}
 	}

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

@@ -7,6 +7,7 @@ export {
 	TaskHandlerRegistry,
 	createDispatchReporter,
 	type TaskHandler,
+	type TaskRun,
 	type DispatchReporter,
 	type DispatchDecision,
 } from './task-handler';

packages/@n8n/scheduler/src/core/executor/lease-constants.ts

@@ -1,5 +1,5 @@
 // These are constants, not environment variables, to keep this mechanism under
-// control in its first version. A per-job timeout may come later.
+// control in its first version.
 
 /**
  * Renewals per lease: two may fail in a row before the lease expires.
@@ -13,10 +13,3 @@ export const RENEWALS_PER_LEASE = 3;
  * renewals, and one this short or shorter expires before its first renewal.
  */
 export const MIN_RENEWAL_INTERVAL_MS = 5_000;
-
-/**
- * A run still pending after this many leases is reported as possibly stuck. Some
- * system tasks run for 15 to 30 minutes, so with the default 60s lease this warns
- * after one hour.
- */
-export const LONG_RUN_THRESHOLD_IN_LEASES = 60;

packages/@n8n/scheduler/src/core/executor/task-handler.ts

@@ -76,20 +76,29 @@ export function createDispatchReporter(onDispatch: () => void): DispatchReporter
  *     after this point is still safe to retry — a redelivery that found the effect
  *     already exists, or nothing done so far.
  *
- * A handler may run as long as it needs: the executor renews the claim's lease
- * while `execute` is pending. `signal` aborts with a `LeaseLostError` when the
+ * A handler may run up to the task's `timeoutSeconds`: the executor renews the
+ * claim's lease while `execute` is pending. `run.signal` aborts with a `LeaseLostError` when the
  * claim is gone, or no renewal succeeded for a whole lease, and no dispatch
  * marker is stored, so another instance may run the same occurrence. Honoring
  * it is optional, but a handler that ignores it may overlap that run. Once
  * `report.dispatched()` has stored the marker, the signal never aborts for a
  * lost claim: no instance runs that occurrence again.
+ *
+ * At the timeout, `run.signal` aborts with a `TaskTimeoutError` and the lease is no
+ * longer renewed, whether or not the handler dispatched. A run that did not
+ * dispatch counts as a failed attempt. A run that dispatched completes. A
+ * handler that returns cleanly after the abort still completes the occurrence.
  */
 export interface TaskHandler {
-	execute(
-		task: ClaimedTask,
-		report: DispatchReporter,
-		signal: AbortSignal,
-	): Promise<DispatchDecision>;
+	execute(task: ClaimedTask, report: DispatchReporter, run: TaskRun): Promise<DispatchDecision>;
+}
+
+/** The lifetime of one handler run. */
+export interface TaskRun {
+	/** Aborts with a `LeaseLostError` or a `TaskTimeoutError`. */
+	readonly signal: AbortSignal;
+	/** Milliseconds until the timeout aborts `signal`. 0 once it did. */
+	remainingMs(): number;
 }
 
 /**

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

@@ -326,11 +326,12 @@ export function createScheduler(deps: SchedulerDeps): Scheduler & SchedulerPasse
 					error: described(error),
 				});
 			},
-			onLongRunningTask: (task, runningSeconds) => {
-				emit('warn', 'Scheduler task is still running after many leases; it may be stuck', {
+			onTaskTimeout: (task) => {
+				recordMetric(() => metrics.recordTaskTimeout(task.taskType));
+				emit('warn', 'Scheduler task run reached its timeout', {
 					taskId: task.id,
 					taskType: task.taskType,
-					runningSeconds,
+					timeoutSeconds: task.timeoutSeconds,
 				});
 			},
 			onDispatch: (taskType, lagSeconds) => {

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

@@ -64,6 +64,7 @@ export { createDispatchReporter, backoff } from './executor';
 export type {
 	ExecutorOptions,
 	TaskHandler,
+	TaskRun,
 	DispatchReporter,
 	DispatchDecision,
 	BackoffOptions,
@@ -112,6 +113,7 @@ export {
 	InvalidOwnerMemberIdError,
 	InvalidRunOptionError,
 	LeaseLostError,
+	TaskTimeoutError,
 } from './errors';
 export type { RetentionOptions, RetentionSummary } from './retention';
 

packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts

@@ -151,12 +151,12 @@ describe('withHandoffTracing', () => {
 		};
 		const task = claimedTask();
 		const report = mock<DispatchReporter>();
-		const signal = new AbortController().signal;
+		const run = { signal: new AbortController().signal, remainingMs: () => 45_000 };
 
-		await withHandoffTracing(tracer, handler).execute(task, report, signal);
+		await withHandoffTracing(tracer, handler).execute(task, report, run);
 
-		// The wrapper adds a span, not semantics: the dispatch reporter and signal flow through.
-		expect(handler.execute).toHaveBeenCalledWith(task, report, signal);
+		// The wrapper adds a span, not semantics: the dispatch reporter and the run flow through.
+		expect(handler.execute).toHaveBeenCalledWith(task, report, run);
 		const options = tracer.startSpan.mock.calls[0][0];
 		expect(options.name).toBe('Scheduler handoff');
 		expect(options.op).toBe('scheduler.handoff');
@@ -173,11 +173,10 @@ describe('withHandoffTracing', () => {
 		const handler: TaskHandler = { execute: vi.fn().mockRejectedValue(new Error('boom')) };
 
 		await expect(
-			withHandoffTracing(tracer, handler).execute(
-				claimedTask(),
-				mock<DispatchReporter>(),
-				new AbortController().signal,
-			),
+			withHandoffTracing(tracer, handler).execute(claimedTask(), mock<DispatchReporter>(), {
+				signal: new AbortController().signal,
+				remainingMs: () => 45_000,
+			}),
 		).rejects.toThrow('boom');
 
 		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatus.error, message: 'boom' });

packages/@n8n/scheduler/src/observability/executor-tracing.ts

@@ -66,9 +66,9 @@ export function createExecutorTracing(tracer: Tracer): ExecutorTracing {
  */
 export function withHandoffTracing(tracer: Tracer, handler: TaskHandler): TaskHandler {
 	return {
-		// The dispatch reporter and signal are threaded through untouched: the wrapper adds
+		// The dispatch reporter and the run are threaded through untouched: the wrapper adds
 		// a span, not semantics, so the handler's dispatch decision keeps flowing to the executor.
-		async execute(task, report, signal) {
+		async execute(task, report, run) {
 			return await tracer.startSpan(
 				{
 					name: 'Scheduler handoff',
@@ -76,7 +76,7 @@ export function withHandoffTracing(tracer: Tracer, handler: TaskHandler): TaskHa
 					attributes: pickSchedulerTaskIdentity(task),
 				},
 				async (span) => {
-					const decision = await handler.execute(task, report, signal);
+					const decision = await handler.execute(task, report, run);
 					span.setStatus({ code: SpanStatus.ok });
 					return decision;
 				},

packages/@n8n/scheduler/src/observability/metrics.ts

@@ -26,6 +26,8 @@ export interface SchedulerMetrics {
 	 * ('lost'), or no renewal succeeded for a whole lease ('expired').
 	 */
 	recordLeaseRenewal(taskType: string, result: LeaseRenewalResult): void;
+	/** A run was stopped because it reached the timeout of its occurrence. */
+	recordTaskTimeout(taskType: string): void;
 
 	/** Outcome of one materialization pass. */
 	recordMaterialized(occurrences: number, deferredJobs: number): void;
@@ -55,6 +57,7 @@ export const noopMetrics: SchedulerMetrics = {
 	observeDispatchLagSeconds() {},
 	recordLeaseLost() {},
 	recordLeaseRenewal() {},
+	recordTaskTimeout() {},
 	recordMaterialized() {},
 	recordMisfired() {},
 	recordRetired() {},

packages/cli/src/metrics/prometheus/__tests__/scheduler-metrics.service.test.ts

@@ -141,6 +141,7 @@ describe('PrometheusSchedulerMetricsService', () => {
 					'n8n_scheduler_jobs_revived_total',
 					'n8n_scheduler_tasks_lease_lost_total',
 					'n8n_scheduler_lease_renewals_total',
+					'n8n_scheduler_task_timeouts_total',
 				]),
 			);
 
@@ -361,6 +362,13 @@ describe('PrometheusSchedulerMetricsService', () => {
 			expect(inc).toHaveBeenCalledWith({ task_type: 'system-task', result: 'lost' }, 1);
 			expect(inc).toHaveBeenCalledTimes(2);
 		});
+
+		it('increments the task-timeout counter by task type', () => {
+			service.recordTaskTimeout('workflow:poll-trigger');
+
+			const inc = counterIncFor('n8n_scheduler_task_timeouts_total');
+			expect(inc).toHaveBeenCalledExactlyOnceWith({ task_type: 'workflow:poll-trigger' }, 1);
+		});
 	});
 
 	describe('push metrics before init', () => {
@@ -381,6 +389,7 @@ describe('PrometheusSchedulerMetricsService', () => {
 			service.recordReconciled(1, 1, 1);
 			service.recordLeaseLost('workflow');
 			service.recordLeaseRenewal('workflow', 'renewed');
+			service.recordTaskTimeout('workflow');
 
 			expect(sharedCounterInc).not.toHaveBeenCalled();
 			expect(mockHistogramObserve).not.toHaveBeenCalled();

packages/cli/src/metrics/prometheus/scheduler-metrics.service.ts

@@ -40,6 +40,7 @@ export class PrometheusSchedulerMetricsService
 	private jobsRevived!: promClient.Counter;
 	private tasksLeaseLost!: promClient.Counter<'task_type'>;
 	private leaseRenewals!: promClient.Counter<'task_type' | 'result'>;
+	private taskTimeouts!: promClient.Counter<'task_type'>;
 	private dispatchLagSeconds!: promClient.Histogram<'task_type'>;
 
 	constructor(
@@ -147,6 +148,12 @@ export class PrometheusSchedulerMetricsService
 			labelNames: ['task_type', 'result'],
 		});
 
+		this.taskTimeouts = new promClient.Counter({
+			name: `${prefix}scheduler_task_timeouts_total`,
+			help: 'Total number of scheduler task runs stopped because they reached their timeout, by task type.',
+			labelNames: ['task_type'],
+		});
+
 		this.dispatchLagSeconds = new promClient.Histogram({
 			name: `${prefix}scheduler_dispatch_lag_seconds`,
 			help: 'Delay in seconds between a task becoming due and being dispatched, by task type.',
@@ -256,6 +263,12 @@ export class PrometheusSchedulerMetricsService
 		}
 	}
 
+	recordTaskTimeout(taskType: string) {
+		if (this.initialized) {
+			this.taskTimeouts.inc({ task_type: taskType }, 1);
+		}
+	}
+
 	recordMaterialized(occurrences: number, deferredJobs: number) {
 		if (this.initialized) {
 			this.occurrencesMaterialized.inc(occurrences);

packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts

@@ -9,6 +9,7 @@ import {
 	type ClaimedTask,
 	type SchedulerMetrics,
 	type SchedulerTaskStore,
+	type TaskRun,
 } from '@n8n/scheduler';
 import type { ErrorReporter, TriggersAndPollers } from 'n8n-core';
 import type { INode, INodeExecutionData, IPollFunctions, IWorkflowBase } from 'n8n-workflow';
@@ -24,6 +25,8 @@ import type { TriggerExecutionContextFactory } from '@/workflows/triggers/trigge
 import { POLL_TRIGGER_TASK_TYPE } from '../poll-trigger-task';
 import { PollTriggerTaskHandler } from '../poll-trigger-task-handler';
 
+const runOf = (signal: AbortSignal): TaskRun => ({ signal, remainingMs: () => 60_000 });
+
 describe('PollTriggerTaskHandler', () => {
 	const nodeTypes = createNodeTypes();
 	const triggerExecutionContextFactory = mock<TriggerExecutionContextFactory>();
@@ -163,7 +166,7 @@ describe('PollTriggerTaskHandler', () => {
 				}),
 			);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggersAndPollers.runPollFunction).not.toHaveBeenCalled();
 			expect(onDispatch).not.toHaveBeenCalled();
@@ -176,7 +179,7 @@ describe('PollTriggerTaskHandler', () => {
 				}),
 			);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggersAndPollers.runPollFunction).not.toHaveBeenCalled();
 			expect(onDispatch).not.toHaveBeenCalled();
@@ -191,7 +194,7 @@ describe('PollTriggerTaskHandler', () => {
 
 	describe('handoff', () => {
 		test('runs poll() against the poll context the factory assembles for the node', async () => {
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggerExecutionContextFactory.createPollExecutionContext).toHaveBeenCalledWith(
 				buildWorkflowData(),
@@ -213,7 +216,7 @@ describe('PollTriggerTaskHandler', () => {
 				backoffUntil: null,
 			});
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggerExecutionContextFactory.createPollExecutionContext).toHaveBeenCalledWith(
 				buildWorkflowData(),
@@ -224,7 +227,7 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('reads workflow data fresh (non-cached) so the poll cursor is never stale', async () => {
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggerExecutionContextFactory.findPublishedWorkflowData).toHaveBeenCalledWith(
 				'wf-1',
@@ -235,7 +238,7 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('hands off and reports a dispatch when poll() returns new data', async () => {
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollFunctions.__emit).toHaveBeenCalledWith(pollData);
 			expect(onDispatch).toHaveBeenCalledTimes(1);
@@ -244,7 +247,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('does not emit and reports no dispatch when poll() returns null', async () => {
 			triggersAndPollers.runPollFunction.mockResolvedValue(null);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollFunctions.__emit).not.toHaveBeenCalled();
 			expect(onDispatch).not.toHaveBeenCalled();
@@ -263,7 +266,7 @@ describe('PollTriggerTaskHandler', () => {
 				return pollResult;
 			});
 
-			await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+			await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(reason);
 
 			expect(pollFunctions.__emit).not.toHaveBeenCalled();
 			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
@@ -288,7 +291,9 @@ describe('PollTriggerTaskHandler', () => {
 					return true;
 				});
 
-				await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+				await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(
+					reason,
+				);
 
 				expect(pollFunctions.__emit).not.toHaveBeenCalled();
 				expect(pollFunctions.__emitError).not.toHaveBeenCalled();
@@ -303,7 +308,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('discards the result and reports no dispatch when the workflow was deactivated during poll()', async () => {
 			workflowRepository.isActive.mockResolvedValue(false);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollFunctions.__emit).not.toHaveBeenCalled();
 			expect(onDispatch).not.toHaveBeenCalled();
@@ -313,7 +318,7 @@ describe('PollTriggerTaskHandler', () => {
 
 	describe('isolate lifecycle', () => {
 		test('acquires the isolate before running poll() and releases it after', async () => {
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(acquireIsolate).toHaveBeenCalledTimes(1);
 			expect(releaseIsolate).toHaveBeenCalledTimes(1);
@@ -326,7 +331,7 @@ describe('PollTriggerTaskHandler', () => {
 			// nothing to release and propagates out for the executor to retry.
 			acquireIsolate.mockRejectedValue(new Error('isolate unavailable'));
 
-			await expect(handler.execute(buildTask(), report, leaseSignal)).rejects.toThrow(
+			await expect(handler.execute(buildTask(), report, runOf(leaseSignal))).rejects.toThrow(
 				'isolate unavailable',
 			);
 
@@ -342,7 +347,7 @@ describe('PollTriggerTaskHandler', () => {
 
 			// Does not rethrow: rethrowing would let the executor retry and re-poll a
 			// still-down source instead of running the error workflow.
-			await expect(handler.execute(buildTask(), report, leaseSignal)).resolves.toBeDefined();
+			await expect(handler.execute(buildTask(), report, runOf(leaseSignal))).resolves.toBeDefined();
 
 			// The cursor is not advanced (no __emit, so no saveStaticData); the error is
 			// handed off to the error workflow via __emitError.
@@ -361,7 +366,7 @@ describe('PollTriggerTaskHandler', () => {
 				throw new Error('poll source unreachable');
 			});
 
-			await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+			await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(reason);
 
 			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
 			expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
@@ -374,7 +379,7 @@ describe('PollTriggerTaskHandler', () => {
 			const commitError = new Error('poller state write failed');
 			pollFunctions.__commitCursor.mockRejectedValue(commitError);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
 			expect(onDispatch).not.toHaveBeenCalled();
@@ -404,7 +409,9 @@ describe('PollTriggerTaskHandler', () => {
 					if (fails) throw new Error('cursor write failed');
 				});
 
-				await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+				await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(
+					reason,
+				);
 
 				expect(pollFunctions.__emitError).not.toHaveBeenCalled();
 				expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
@@ -440,7 +447,7 @@ describe('PollTriggerTaskHandler', () => {
 			else triggersAndPollers.runPollFunction.mockResolvedValue(poll);
 			workflowRepository.isActive.mockResolvedValue(active);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollFunctions.__commitCursor).toHaveBeenCalledTimes(commits);
 		});
@@ -462,7 +469,7 @@ describe('PollTriggerTaskHandler', () => {
 				scopeAtCommit = scope.getStore();
 			});
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(scopeAtCommit).toBe('staging');
 		});
@@ -473,7 +480,7 @@ describe('PollTriggerTaskHandler', () => {
 				scopeAtEmit = scope.getStore();
 			});
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(scopeAtEmit).toBe('staging');
 		});
@@ -511,9 +518,9 @@ describe('PollTriggerTaskHandler', () => {
 					});
 				}
 
-				const rejected = expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(
-					reason,
-				);
+				const rejected = expect(
+					handler.execute(buildTask(), report, runOf(lease.signal)),
+				).rejects.toBe(reason);
 				await vi.advanceTimersByTimeAsync(pollTimeoutMs);
 				await rejected;
 
@@ -534,9 +541,9 @@ describe('PollTriggerTaskHandler', () => {
 			const reason = new Error('lease expired');
 			triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
 
-			const rejected = expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(
-				reason,
-			);
+			const rejected = expect(
+				handler.execute(buildTask(), report, runOf(lease.signal)),
+			).rejects.toBe(reason);
 			await vi.advanceTimersByTimeAsync(1);
 			lease.abort(reason);
 			await rejected;
@@ -551,7 +558,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('abandons a poll that outlives the timeout and reports no dispatch', async () => {
 			triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
 
-			const executing = handler.execute(buildTask(), report, leaseSignal);
+			const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
 			await vi.advanceTimersByTimeAsync(pollTimeoutMs);
 
 			await expect(executing).resolves.toBeDefined();
@@ -580,7 +587,7 @@ describe('PollTriggerTaskHandler', () => {
 			workflowRepository.isActive.mockResolvedValue(false);
 			triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
 
-			const executing = handler.execute(buildTask(), report, leaseSignal);
+			const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
 			await vi.advanceTimersByTimeAsync(pollTimeoutMs);
 			await executing;
 
@@ -595,7 +602,7 @@ describe('PollTriggerTaskHandler', () => {
 				}),
 			);
 
-			const executing = handler.execute(buildTask(), report, leaseSignal);
+			const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
 			await vi.advanceTimersByTimeAsync(pollTimeoutMs - 1);
 			resolvePoll(pollData);
 			await executing;
@@ -616,7 +623,7 @@ describe('PollTriggerTaskHandler', () => {
 				}),
 			);
 
-			const executing = handler.execute(buildTask(), report, leaseSignal);
+			const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
 			await vi.advanceTimersByTimeAsync(pollTimeoutMs);
 			await executing;
 			resolvePoll(pollData);
@@ -646,7 +653,7 @@ describe('PollTriggerTaskHandler', () => {
 				globalConfig,
 			);
 
-			const executing = plainHandler.execute(buildTask(), report, leaseSignal);
+			const executing = plainHandler.execute(buildTask(), report, runOf(leaseSignal));
 			await vi.advanceTimersByTimeAsync(pollTimeoutMs);
 			await executing;
 			rejectPoll(new Error('poll source unreachable'));
@@ -726,7 +733,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('rejects a task whose payload is missing workflowId or nodeId', async () => {
 			const task = buildTask({ payload: { nodeId: 'node-1' } });
 
-			await expect(handler.execute(task, report, leaseSignal)).rejects.toThrow(
+			await expect(handler.execute(task, report, runOf(leaseSignal))).rejects.toThrow(
 				'Poll-trigger task payload is missing workflowId or nodeId',
 			);
 			expect(triggerExecutionContextFactory.findPublishedWorkflowData).not.toHaveBeenCalled();
@@ -736,7 +743,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('reports no dispatch when the published workflow is gone', async () => {
 			triggerExecutionContextFactory.findPublishedWorkflowData.mockResolvedValue(null);
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			// The workflow was unpublished after the claim: the occurrence completes as a
 			// no-op instead of retrying to dead-letter while the owner retires the job.
@@ -757,7 +764,7 @@ describe('PollTriggerTaskHandler', () => {
 				buildWorkflowData({ nodes }),
 			);
 
-			await expect(handler.execute(buildTask(), report, leaseSignal)).rejects.toThrow(
+			await expect(handler.execute(buildTask(), report, runOf(leaseSignal))).rejects.toThrow(
 				'missing or disabled in the published workflow',
 			);
 			expect(triggersAndPollers.runPollFunction).not.toHaveBeenCalled();
@@ -770,7 +777,7 @@ describe('PollTriggerTaskHandler', () => {
 			(pollFunctions as IPollFunctions).__commitCursor = undefined;
 			triggersAndPollers.runPollFunction.mockResolvedValue(null);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggersAndPollers.runPollFunction).toHaveBeenCalledWith(
 				workflow,
@@ -799,7 +806,7 @@ describe('PollTriggerTaskHandler', () => {
 				lease.abort(reason);
 			});
 
-			await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+			await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(reason);
 
 			expect(pollFunctions.__emit).not.toHaveBeenCalled();
 			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
@@ -827,7 +834,9 @@ describe('PollTriggerTaskHandler', () => {
 					});
 				}
 
-				await expect(handler.execute(buildTask(), report, lease.signal)).rejects.toBe(reason);
+				await expect(handler.execute(buildTask(), report, runOf(lease.signal))).rejects.toBe(
+					reason,
+				);
 
 				expect(pollBackoffService.recordFailure).toHaveBeenCalledTimes(
 					abortDuring === 'failure write' ? 1 : 0,
@@ -847,7 +856,7 @@ describe('PollTriggerTaskHandler', () => {
 			pollBackoffService.getState.mockResolvedValue(state);
 			pollBackoffService.isBackingOff.mockReturnValue(true);
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(decision).toBe(report.notDispatched());
 			expect(triggerExecutionContextFactory.findPublishedWorkflowData).not.toHaveBeenCalled();
@@ -862,7 +871,7 @@ describe('PollTriggerTaskHandler', () => {
 			const error = new Error('poll source unreachable');
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordFailure).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -881,7 +890,7 @@ describe('PollTriggerTaskHandler', () => {
 				new UnexpectedError('Node type does not have a poll function defined'),
 			);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordFailure).toHaveBeenCalledTimes(1);
 		});
@@ -892,7 +901,7 @@ describe('PollTriggerTaskHandler', () => {
 			const error = new Error('database unavailable');
 			workflowRepository.isActive.mockRejectedValue(error);
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggersAndPollers.runPollFunction).toHaveBeenCalled();
 			expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
@@ -906,7 +915,7 @@ describe('PollTriggerTaskHandler', () => {
 			triggersAndPollers.runPollFunction.mockResolvedValue(null);
 			pollFunctions.__commitCursor.mockRejectedValue(new Error('poller state write failed'));
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordSuccess).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -923,7 +932,7 @@ describe('PollTriggerTaskHandler', () => {
 			pollBackoffService.getState.mockResolvedValue(state);
 			triggersAndPollers.runPollFunction.mockResolvedValue(pollResult);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordSuccess).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -938,7 +947,7 @@ describe('PollTriggerTaskHandler', () => {
 			pollBackoffService.getState.mockResolvedValue(state);
 			workflowRepository.isActive.mockResolvedValue(false);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordSuccess).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -954,7 +963,7 @@ describe('PollTriggerTaskHandler', () => {
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 			workflowRepository.isActive.mockResolvedValue(false);
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
 			expect(pollFunctions.__emitError).toHaveBeenCalledWith(error);
@@ -968,7 +977,7 @@ describe('PollTriggerTaskHandler', () => {
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 			workflowRepository.isActive.mockRejectedValue(new Error('database unavailable'));
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(pollBackoffService.recordFailure).toHaveBeenCalledWith({
 				workflowId: 'wf-1',
@@ -984,7 +993,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('does not touch the failure counters when the published workflow is missing', async () => {
 			triggerExecutionContextFactory.findPublishedWorkflowData.mockResolvedValue(null);
 
-			const decision = await handler.execute(buildTask(), report, leaseSignal);
+			const decision = await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(decision).toBe(report.notDispatched());
 			expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
@@ -994,7 +1003,7 @@ describe('PollTriggerTaskHandler', () => {
 		test('does not read the failure state when the payload is invalid', async () => {
 			const task = buildTask({ payload: { nodeId: 'node-1' } });
 
-			await expect(handler.execute(task, report, leaseSignal)).rejects.toThrow();
+			await expect(handler.execute(task, report, runOf(leaseSignal))).rejects.toThrow();
 
 			expect(pollBackoffService.getState).not.toHaveBeenCalled();
 		});
@@ -1003,7 +1012,7 @@ describe('PollTriggerTaskHandler', () => {
 			const error = new Error('poller state read failed');
 			pollBackoffService.getState.mockRejectedValue(error);
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			expect(triggersAndPollers.runPollFunction).toHaveBeenCalled();
 			expect(onDispatch).toHaveBeenCalledTimes(1);
@@ -1016,7 +1025,7 @@ describe('PollTriggerTaskHandler', () => {
 				throw error;
 			});
 
-			await handler.execute(buildTask(), report, leaseSignal);
+			await handler.execute(buildTask(), report, runOf(leaseSignal));
 
 			const [, isBackingOffNow] = pollBackoffService.isBackingOff.mock.calls[0];
 			const { now: recordFailureNow } = pollBackoffService.recordFailure.mock.calls[0][0];

packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts

@@ -5,7 +5,13 @@ import { Time } from '@n8n/constants';
 import type { PollerFailureState } from '@n8n/db';
 import { WorkflowRepository } from '@n8n/db';
 import { Service } from '@n8n/di';
-import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler';
+import type {
+	ClaimedTask,
+	DispatchDecision,
+	DispatchReporter,
+	TaskHandler,
+	TaskRun,
+} from '@n8n/scheduler';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import {
 	commitStagedCursor,
@@ -64,7 +70,7 @@ export class PollTriggerTaskHandler implements TaskHandler {
 	async execute(
 		task: ClaimedTask,
 		report: DispatchReporter,
-		leaseSignal: AbortSignal,
+		{ signal: leaseSignal }: TaskRun,
 	): Promise<DispatchDecision> {
 		// A setup failure here retries to N8N_SCHEDULER_MAX_ATTEMPTS then dead-letters,
 		// unlike a `poll()` runtime failure below, which routes to the error workflow instead.

packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts

@@ -1,6 +1,6 @@
 import type { EventService } from '@n8n/backend-services';
 import { mockLogger } from '@n8n/backend-test-utils';
-import type { ClaimedTask, DispatchReporter } from '@n8n/scheduler';
+import type { ClaimedTask, DispatchReporter, TaskRun } from '@n8n/scheduler';
 import { createDispatchReporter, LeaseLostError } from '@n8n/scheduler';
 import { Tracing } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
@@ -8,6 +8,8 @@ import { mock } from 'vitest-mock-extended';
 import { SystemTaskHandler } from '../system-task-handler';
 import { DummySystemTask } from './dummy.task';
 
+const runOf = (signal: AbortSignal): TaskRun => ({ signal, remainingMs: () => 60_000 });
+
 describe('SystemTaskHandler', () => {
 	const claimed = mock<ClaimedTask>({ id: 'task-1', jobId: 2 });
 
@@ -41,7 +43,7 @@ describe('SystemTaskHandler', () => {
 	it('runs the task', async () => {
 		const { task, report, handler } = setup('idempotent');
 
-		await handler.execute(claimed, report, new AbortController().signal);
+		await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 		expect(task.runCount).toBe(1);
 	});
@@ -55,7 +57,7 @@ describe('SystemTaskHandler', () => {
 				seenSignal = signal;
 			};
 
-			await context.handler.execute(claimed, context.report, context.leaseController.signal);
+			await context.handler.execute(claimed, context.report, runOf(context.leaseController.signal));
 			expect(seenSignal?.aborted).toBe(false);
 			context[source].abort();
 
@@ -70,7 +72,7 @@ describe('SystemTaskHandler', () => {
 				signal.addEventListener('abort', () => reject(new Error('aborted')));
 			});
 
-		const executing = handler.execute(claimed, report, leaseController.signal);
+		const executing = handler.execute(claimed, report, runOf(leaseController.signal));
 		leaseController.abort();
 
 		await expect(executing).rejects.toThrow();
@@ -86,7 +88,7 @@ describe('SystemTaskHandler', () => {
 			});
 		const error = new LeaseLostError();
 
-		const executing = handler.execute(claimed, report, leaseController.signal);
+		const executing = handler.execute(claimed, report, runOf(leaseController.signal));
 		leaseController.abort(error);
 
 		await expect(executing).rejects.toBe(error);
@@ -107,7 +109,7 @@ describe('SystemTaskHandler', () => {
 				runCountsAtDispatch.push(task.runCount);
 			});
 
-			await handler.execute(claimed, report, new AbortController().signal);
+			await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 			expect(runCountsAtDispatch).toEqual([1]);
 		},
@@ -128,7 +130,7 @@ describe('SystemTaskHandler', () => {
 				vi.fn(),
 			);
 
-			const returned = await handler.execute(claimed, report, new AbortController().signal);
+			const returned = await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 			expect(returned).toBe(report.dispatched());
 		},
@@ -140,9 +142,9 @@ describe('SystemTaskHandler', () => {
 			throw new Error('failed');
 		};
 
-		await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
-			'failed',
-		);
+		await expect(
+			handler.execute(claimed, report, runOf(new AbortController().signal)),
+		).rejects.toThrow('failed');
 	});
 
 	it.each(['idempotent', 'non-idempotent'] as const)(
@@ -154,9 +156,9 @@ describe('SystemTaskHandler', () => {
 				throw error;
 			};
 
-			await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
-				error,
-			);
+			await expect(
+				handler.execute(claimed, report, runOf(new AbortController().signal)),
+			).rejects.toThrow(error);
 
 			expect(onRunError).toHaveBeenCalledWith(error);
 		},
@@ -169,7 +171,7 @@ describe('SystemTaskHandler', () => {
 				signal.addEventListener('abort', () => reject(new Error('aborted')));
 			});
 
-		const executing = handler.execute(claimed, report, new AbortController().signal);
+		const executing = handler.execute(claimed, report, runOf(new AbortController().signal));
 		shutdownController.abort();
 
 		await expect(executing).rejects.toThrow();
@@ -182,7 +184,7 @@ describe('SystemTaskHandler', () => {
 			throw new Error('sink');
 		});
 
-		await handler.execute(claimed, report, new AbortController().signal);
+		await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 		expect(task.runCount).toBe(1);
 		expect(onRunError).not.toHaveBeenCalled();
@@ -191,7 +193,7 @@ describe('SystemTaskHandler', () => {
 	it('does not report a run that succeeds', async () => {
 		const { report, handler, onRunError } = setup('idempotent');
 
-		await handler.execute(claimed, report, new AbortController().signal);
+		await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 		expect(onRunError).not.toHaveBeenCalled();
 	});
@@ -211,7 +213,7 @@ describe('SystemTaskHandler', () => {
 				await vi.advanceTimersByTimeAsync(250);
 			};
 
-			await handler.execute(claimed, report, new AbortController().signal);
+			await handler.execute(claimed, report, runOf(new AbortController().signal));
 
 			expect(eventService.emit.mock.calls).toEqual([
 				['system-task-run-started', { name: 'dummy', mode: 'durable' }],
@@ -228,9 +230,9 @@ describe('SystemTaskHandler', () => {
 				throw new Error('failed');
 			};
 
-			await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
-				'failed',
-			);
+			await expect(
+				handler.execute(claimed, report, runOf(new AbortController().signal)),
+			).rejects.toThrow('failed');
 
 			expect(eventService.emit).toHaveBeenCalledWith(
 				'system-task-run-settled',
@@ -245,7 +247,7 @@ describe('SystemTaskHandler', () => {
 					signal.addEventListener('abort', () => resolve());
 				});
 
-			const executing = handler.execute(claimed, report, new AbortController().signal);
+			const executing = handler.execute(claimed, report, runOf(new AbortController().signal));
 			shutdownController.abort();
 			await executing;
 
@@ -262,7 +264,7 @@ describe('SystemTaskHandler', () => {
 					signal.addEventListener('abort', () => reject(new Error('aborted')));
 				});
 
-			const executing = handler.execute(claimed, report, new AbortController().signal);
+			const executing = handler.execute(claimed, report, runOf(new AbortController().signal));
 			shutdownController.abort();
 			await expect(executing).rejects.toThrow();
 
@@ -279,7 +281,7 @@ describe('SystemTaskHandler', () => {
 					signal.addEventListener('abort', () => reject(new Error('aborted')));
 				});
 
-			const executing = handler.execute(claimed, report, leaseController.signal);
+			const executing = handler.execute(claimed, report, runOf(leaseController.signal));
 			leaseController.abort();
 			await expect(executing).rejects.toThrow();
 

packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts

@@ -5,7 +5,7 @@ import { Time } from '@n8n/constants';
 import type { ScheduledJobRepository } from '@n8n/db';
 import { SystemTaskMetadata } from '@n8n/decorators';
 import { Container } from '@n8n/di';
-import type { ClaimedTask } from '@n8n/scheduler';
+import type { ClaimedTask, TaskRun } from '@n8n/scheduler';
 import { createDispatchReporter } from '@n8n/scheduler';
 import { Tracing, type ErrorReporter, type InstanceSettings } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
@@ -20,6 +20,8 @@ import { DummySystemTask, OtherDummySystemTask, PerInstanceDummySystemTask } fro
 const START = new Date('2026-01-01T00:00:00.000Z');
 const ONE_INTERVAL_MS = 60 * Time.seconds.toMilliseconds;
 
+const runOf = (signal: AbortSignal): TaskRun => ({ signal, remainingMs: () => 60_000 });
+
 async function initRunner(runner: SystemTaskRunner): Promise<void> {
 	await runner.init();
 }
@@ -790,7 +792,7 @@ describe('SystemTaskRunner', () => {
 				handler.execute(
 					mock<ClaimedTask>(),
 					createDispatchReporter(vi.fn()),
-					new AbortController().signal,
+					runOf(new AbortController().signal),
 				),
 			).rejects.toThrow(error);
 
@@ -819,7 +821,7 @@ describe('SystemTaskRunner', () => {
 			const executing = handler.execute(
 				mock<ClaimedTask>(),
 				createDispatchReporter(vi.fn()),
-				new AbortController().signal,
+				runOf(new AbortController().signal),
 			);
 			expect(runSignal?.aborted).toBe(false);
 
@@ -1342,7 +1344,7 @@ describe('SystemTaskRunner', () => {
 			await handler.execute(
 				mock<ClaimedTask>(),
 				createDispatchReporter(vi.fn()),
-				new AbortController().signal,
+				runOf(new AbortController().signal),
 			);
 
 			expect(eventService.emit).toHaveBeenCalledWith('system-task-run-started', {

packages/cli/src/scheduling/system-tasks/system-task-handler.ts

@@ -1,7 +1,13 @@
 import type { Logger } from '@n8n/backend-common';
 import type { EventService } from '@n8n/backend-services';
 import type { SystemTask } from '@n8n/decorators';
-import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler';
+import type {
+	ClaimedTask,
+	DispatchDecision,
+	DispatchReporter,
+	TaskHandler,
+	TaskRun,
+} from '@n8n/scheduler';
 import type { Tracing } from 'n8n-core';
 
 import { observeSystemTaskRun } from './system-task-run-observer';
@@ -26,7 +32,7 @@ export class SystemTaskHandler implements TaskHandler {
 	async execute(
 		task: ClaimedTask,
 		report: DispatchReporter,
-		leaseSignal: AbortSignal,
+		{ signal: leaseSignal }: TaskRun,
 	): Promise<DispatchDecision> {
 		const outcome = await observeSystemTaskRun(
 			this.eventService,

packages/cli/test/integration/scheduling/scheduler-task-timeout.test.ts

@@ -0,0 +1,104 @@
+import { testDb } from '@n8n/backend-test-utils';
+import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { createScheduler } from '@n8n/scheduler';
+import type { Scheduler, SchedulerMetrics, SchedulerPasses } from '@n8n/scheduler';
+import { mock } from 'vitest-mock-extended';
+
+import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
+
+import { selfOwned } from './shared/job-factory';
+
+/**
+ * A handler that never settles and ignores its signal stops holding its claim
+ * once it reaches its timeout: the executor stops renewing the lease, so the
+ * reaper recovers the occurrence within one lease after the timeout.
+ */
+describe('scheduler task timeout over the storage bindings', () => {
+	const TASK_TYPE = 'integration-task-timeout-test';
+	// The first renewal lands after five seconds, before the timeout, so the lease
+	// outlives its first term only through the heartbeat.
+	const LEASE_SECONDS = 6;
+	const TIMEOUT_SECONDS = 7;
+
+	let jobRepo: ScheduledJobRepository;
+	let taskRepo: ScheduledTaskRepository;
+	let scheduler: Scheduler & SchedulerPasses;
+	const metrics = mock<SchedulerMetrics>();
+	let signal: AbortSignal | undefined;
+
+	beforeAll(async () => {
+		await testDb.init();
+		jobRepo = Container.get(ScheduledJobRepository);
+		taskRepo = Container.get(ScheduledTaskRepository);
+		scheduler = createScheduler({
+			hostId: 'main-task-timeout-test',
+			materializerTransaction: buildMaterializerTransaction(
+				Container.get(DataSource),
+				jobRepo,
+				taskRepo,
+			),
+			taskStore: taskRepo,
+			executor: { leaseSeconds: LEASE_SECONDS, lookaheadSeconds: 1 },
+			metrics,
+		});
+		scheduler.registerTaskHandler(TASK_TYPE, {
+			execute: async (_task, _report, run) => {
+				signal = run.signal;
+				return await new Promise<never>(() => {});
+			},
+		});
+	});
+
+	afterAll(async () => {
+		await scheduler.stop();
+		await testDb.terminate();
+	});
+
+	it('recovers a handler that never settles after its timeout plus one lease', async () => {
+		const job = await jobRepo.save(
+			jobRepo.create({
+				name: 'job-task-timeout',
+				...selfOwned('job-task-timeout'),
+				taskType: TASK_TYPE,
+				payload: {},
+				kind: 'interval',
+				intervalSeconds: 3600,
+				enabled: true,
+				nextRunAt: new Date(Date.now() - 1000),
+				maxAttempts: 3,
+				timeoutSeconds: TIMEOUT_SECONDS,
+			}),
+		);
+		await scheduler.materialize();
+		const [claimed] = await scheduler.execute();
+		expect(claimed.timeoutSeconds).toBe(TIMEOUT_SECONDS);
+		const dispatchedAt = Date.now();
+
+		// Sweep like a busy reaper until the occurrence comes back.
+		const deadline = dispatchedAt + (TIMEOUT_SECONDS + LEASE_SECONDS + 5) * 1_000;
+		let reclaimedAt: number | undefined;
+		while (reclaimedAt === undefined && Date.now() < deadline) {
+			const { reclaimed } = await scheduler.reap();
+			if (reclaimed > 0) {
+				reclaimedAt = Date.now();
+			} else {
+				await new Promise((resolve) => setTimeout(resolve, 200));
+			}
+		}
+
+		expect(reclaimedAt).toBeDefined();
+		const elapsedMs = reclaimedAt! - dispatchedAt;
+		expect(elapsedMs).toBeGreaterThanOrEqual(TIMEOUT_SECONDS * 1_000);
+		expect(elapsedMs).toBeLessThanOrEqual((TIMEOUT_SECONDS + LEASE_SECONDS + 1) * 1_000);
+		expect(signal?.reason).toMatchObject({
+			name: 'TaskTimeoutError',
+			timeoutSeconds: TIMEOUT_SECONDS,
+		});
+		expect(metrics.recordTaskTimeout).toHaveBeenCalledExactlyOnceWith(TASK_TYPE);
+
+		const task = await taskRepo.findOneByOrFail({ jobId: job.id });
+		expect(task.status).toBe('pending');
+		expect(task.attempts).toBe(1);
+	}, 30_000);
+});