feat(core): Stop a durable system task run when its claim is lost (no-changelog) (#40272)

packages/@n8n/decorators/src/system-task/system-task.ts

@@ -74,10 +74,12 @@ export interface SystemTask {
 	readonly concurrencyLimit?: number | null;
 
 	/**
-	 * Executes one occurrence of the task. `signal` aborts when the run should
-	 * stop early: the instance is shutting down or, for a run on the in-memory
-	 * timer, leadership was lost. Honoring it is optional, but a run that
-	 * ignores it delays stepdown and shutdown until it settles.
+	 * Executes one occurrence of the task. A run may take as long as it needs.
+	 * `signal` aborts on shutdown, and on loss of leadership for an in-memory timer.
+	 * A durable run also aborts on lease loss or expiry, unless its dispatch marker is stored.
+	 * Ignoring the signal delays shutdown and can let another run overlap.
+	 * An idempotent durable run that settles after a lease abort counts as a failed
+	 * attempt while its claim still matches, and retries only while attempts remain.
 	 */
 	run(signal: AbortSignal): Promise<void>;
 }

packages/@n8n/scheduler/README.md

@@ -407,8 +407,10 @@ The design leans on a few ideas working together.
   expires before its first renewal, and the scheduler warns about it at startup.
   If a renewal finds the claim gone, or no renewal succeeds for a whole lease, the
   handler's signal aborts, unless the run is already recorded as dispatched.
-  A handler that stops on this abort uses up an attempt, as when the reaper takes
-  back an expired lease, even if a later renewal kept the claim.
+  A handler that abandons work on this abort must reject. If its claim still
+  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.
 - **Fencing.** Each claim carries a version number (an *epoch*) that increases every

packages/cli/src/events/maps/system-task-metrics.event-map.ts

@@ -14,9 +14,10 @@ export type SystemTaskMode = 'leader_timer' | 'instance_timer' | 'durable';
 
 /**
  * How a run settled. A rejection after the run's abort signal fired is the task
- * honoring the abort, so it is `aborted`, not `failure`.
+ * honoring the abort, so it is `aborted`, not `failure`. A durable run stopped
+ * because its claim was lost is `lease_lost`.
  */
-export type SystemTaskRunResult = 'success' | 'failure' | 'aborted';
+export type SystemTaskRunResult = 'success' | 'failure' | 'aborted' | 'lease_lost';
 
 /**
  * Why an occurrence did not run: the previous run was still going, another

packages/cli/src/metrics/prometheus/system-task-metrics.service.ts

@@ -65,7 +65,7 @@ export class PrometheusSystemTaskMetricsService implements PrometheusMetricsColl
 
 		const runDuration = new promClient.Histogram({
 			name: `${prefix}system_task_run_duration_seconds`,
-			help: 'Duration in seconds of a system task run, by task, mode (leader_timer, instance_timer, durable) and result (success, failure, aborted).',
+			help: 'Duration in seconds of a system task run, by task, mode (leader_timer, instance_timer, durable) and result (success, failure, aborted, lease_lost).',
 			labelNames: ['task', 'mode', 'result'],
 			buckets: DURATION_BUCKETS_SECONDS,
 		});

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

@@ -1,7 +1,7 @@
 import type { EventService } from '@n8n/backend-services';
 import { mockLogger } from '@n8n/backend-test-utils';
 import type { ClaimedTask, DispatchReporter } from '@n8n/scheduler';
-import { createDispatchReporter } from '@n8n/scheduler';
+import { createDispatchReporter, LeaseLostError } from '@n8n/scheduler';
 import { Tracing } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
 
@@ -18,6 +18,7 @@ describe('SystemTaskHandler', () => {
 		const onRunError = vi.fn();
 		const eventService = mock<EventService>();
 		const shutdownController = new AbortController();
+		const leaseController = new AbortController();
 		const handler = new SystemTaskHandler(
 			task,
 			shutdownController.signal,
@@ -33,34 +34,74 @@ describe('SystemTaskHandler', () => {
 			onRunError,
 			eventService,
 			shutdownController,
-			shutdownSignal: shutdownController.signal,
+			leaseController,
 		};
 	}
 
 	it('runs the task', async () => {
 		const { task, report, handler } = setup('idempotent');
 
-		await handler.execute(claimed, report);
+		await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(task.runCount).toBe(1);
 	});
 
-	it('hands the shutdown signal to the run', async () => {
-		const { task, report, handler, shutdownSignal } = setup('idempotent');
-		let seenSignal: AbortSignal | undefined;
-		task.onRun = async (signal) => {
-			seenSignal = signal;
-		};
+	it.each(['shutdownController', 'leaseController'] as const)(
+		'hands the run a signal that aborts with the %s',
+		async (source) => {
+			const context = setup('idempotent');
+			let seenSignal: AbortSignal | undefined;
+			context.task.onRun = async (signal) => {
+				seenSignal = signal;
+			};
 
-		await handler.execute(claimed, report);
+			await context.handler.execute(claimed, context.report, context.leaseController.signal);
+			expect(seenSignal?.aborted).toBe(false);
+			context[source].abort();
 
-		expect(seenSignal).toBe(shutdownSignal);
+			expect(seenSignal?.aborted).toBe(true);
+		},
+	);
+
+	it('does not report a run that rejects once the claim was lost', async () => {
+		const { task, report, handler, onRunError, leaseController } = setup('idempotent');
+		task.onRun = async (signal) =>
+			await new Promise<void>((_, reject) => {
+				signal.addEventListener('abort', () => reject(new Error('aborted')));
+			});
+
+		const executing = handler.execute(claimed, report, leaseController.signal);
+		leaseController.abort();
+
+		await expect(executing).rejects.toThrow();
+		expect(onRunError).not.toHaveBeenCalled();
+	});
+
+	it('propagates the lease error when the run resolves after its claim is lost', async () => {
+		const { task, report, handler, onRunError, eventService, leaseController } =
+			setup('idempotent');
+		task.onRun = async (signal) =>
+			await new Promise<void>((resolve) => {
+				signal.addEventListener('abort', () => resolve(), { once: true });
+			});
+		const error = new LeaseLostError();
+
+		const executing = handler.execute(claimed, report, leaseController.signal);
+		leaseController.abort(error);
+
+		await expect(executing).rejects.toBe(error);
+		expect(report.dispatched).not.toHaveBeenCalled();
+		expect(onRunError).not.toHaveBeenCalled();
+		expect(eventService.emit).toHaveBeenCalledWith(
+			'system-task-run-settled',
+			expect.objectContaining({ mode: 'durable', result: 'lease_lost' }),
+		);
 	});
 
 	it('leaves idempotent work retryable', async () => {
 		const { report, handler } = setup('idempotent');
 
-		await handler.execute(claimed, report);
+		await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(report.notDispatched).toHaveBeenCalled();
 		expect(report.dispatched).not.toHaveBeenCalled();
@@ -72,7 +113,7 @@ describe('SystemTaskHandler', () => {
 			expect(report.dispatched).toHaveBeenCalled();
 		};
 
-		await handler.execute(claimed, report);
+		await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(task.runCount).toBe(1);
 		expect(report.notDispatched).not.toHaveBeenCalled();
@@ -94,7 +135,7 @@ describe('SystemTaskHandler', () => {
 			vi.fn(),
 		);
 
-		const returned = await handler.execute(claimed, report);
+		const returned = await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(returned).toBe(report[decision]());
 	});
@@ -105,7 +146,9 @@ describe('SystemTaskHandler', () => {
 			throw new Error('failed');
 		};
 
-		await expect(handler.execute(claimed, report)).rejects.toThrow('failed');
+		await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
+			'failed',
+		);
 	});
 
 	it.each(['idempotent', 'non-idempotent'] as const)(
@@ -117,7 +160,9 @@ describe('SystemTaskHandler', () => {
 				throw error;
 			};
 
-			await expect(handler.execute(claimed, report)).rejects.toThrow(error);
+			await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
+				error,
+			);
 
 			expect(onRunError).toHaveBeenCalledWith(error);
 		},
@@ -130,7 +175,7 @@ describe('SystemTaskHandler', () => {
 				signal.addEventListener('abort', () => reject(new Error('aborted')));
 			});
 
-		const executing = handler.execute(claimed, report);
+		const executing = handler.execute(claimed, report, new AbortController().signal);
 		shutdownController.abort();
 
 		await expect(executing).rejects.toThrow();
@@ -143,7 +188,7 @@ describe('SystemTaskHandler', () => {
 			throw new Error('sink');
 		});
 
-		await handler.execute(claimed, report);
+		await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(task.runCount).toBe(1);
 		expect(onRunError).not.toHaveBeenCalled();
@@ -152,7 +197,7 @@ describe('SystemTaskHandler', () => {
 	it('does not report a run that succeeds', async () => {
 		const { report, handler, onRunError } = setup('idempotent');
 
-		await handler.execute(claimed, report);
+		await handler.execute(claimed, report, new AbortController().signal);
 
 		expect(onRunError).not.toHaveBeenCalled();
 	});
@@ -172,7 +217,7 @@ describe('SystemTaskHandler', () => {
 				await vi.advanceTimersByTimeAsync(250);
 			};
 
-			await handler.execute(claimed, report);
+			await handler.execute(claimed, report, new AbortController().signal);
 
 			expect(eventService.emit.mock.calls).toEqual([
 				['system-task-run-started', { name: 'dummy', mode: 'durable' }],
@@ -189,7 +234,9 @@ describe('SystemTaskHandler', () => {
 				throw new Error('failed');
 			};
 
-			await expect(handler.execute(claimed, report)).rejects.toThrow('failed');
+			await expect(handler.execute(claimed, report, new AbortController().signal)).rejects.toThrow(
+				'failed',
+			);
 
 			expect(eventService.emit).toHaveBeenCalledWith(
 				'system-task-run-settled',
@@ -204,7 +251,7 @@ describe('SystemTaskHandler', () => {
 					signal.addEventListener('abort', () => resolve());
 				});
 
-			const executing = handler.execute(claimed, report);
+			const executing = handler.execute(claimed, report, new AbortController().signal);
 			shutdownController.abort();
 			await executing;
 
@@ -221,7 +268,7 @@ describe('SystemTaskHandler', () => {
 					signal.addEventListener('abort', () => reject(new Error('aborted')));
 				});
 
-			const executing = handler.execute(claimed, report);
+			const executing = handler.execute(claimed, report, new AbortController().signal);
 			shutdownController.abort();
 			await expect(executing).rejects.toThrow();
 
@@ -230,5 +277,22 @@ describe('SystemTaskHandler', () => {
 				expect.objectContaining({ mode: 'durable', result: 'aborted' }),
 			);
 		});
+
+		it('settles a run that rejects once its claim was lost as lease_lost', async () => {
+			const { task, report, handler, eventService, leaseController } = setup('idempotent');
+			task.onRun = async (signal) =>
+				await new Promise<void>((_, reject) => {
+					signal.addEventListener('abort', () => reject(new Error('aborted')));
+				});
+
+			const executing = handler.execute(claimed, report, leaseController.signal);
+			leaseController.abort();
+			await expect(executing).rejects.toThrow();
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'system-task-run-settled',
+				expect.objectContaining({ mode: 'durable', result: 'lease_lost' }),
+			);
+		});
 	});
 });

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

@@ -167,4 +167,31 @@ describe('observeSystemTaskRun', () => {
 		expect(span.setAttribute).toHaveBeenCalledWith('n8n.system_task.result', 'aborted');
 		expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatus.ok });
 	});
+
+	it.each([
+		{ aborted: ['lease'], result: 'lease_lost' },
+		{ aborted: ['shutdown'], result: 'aborted' },
+		{ aborted: ['shutdown', 'lease'], result: 'lease_lost' },
+	] as const)('settles a run stopped by $aborted as $result', async ({ aborted, result }) => {
+		const eventService = mock<EventService>();
+		const controllers = { shutdown: new AbortController(), lease: new AbortController() };
+		const task = taskThat(async () => {
+			aborted.forEach((source) => controllers[source].abort());
+		});
+
+		const outcome = await observeSystemTaskRun(
+			eventService,
+			setupTracing().tracing,
+			task,
+			'durable',
+			controllers.shutdown.signal,
+			controllers.lease.signal,
+		);
+
+		expect(outcome.result).toBe(result);
+		expect(eventService.emit).toHaveBeenCalledWith(
+			'system-task-run-settled',
+			expect.objectContaining({ result }),
+		);
+	});
 });

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

@@ -7,8 +7,8 @@ import type { Tracing } from 'n8n-core';
 import { observeSystemTaskRun } from './system-task-run-observer';
 
 /**
- * Runs one durable occurrence of a system task, handing it `shutdownSignal` so
- * it can stop early when the instance shuts down.
+ * Runs one durable occurrence of a system task. The signal aborts on shutdown.
+ * Lease loss or expiry also aborts it unless the dispatch marker is stored.
  *
  * Errors propagate: the executor is what retries the occurrence or gives up on
  * it, following the attempt limit carried by the occurrence's job row.
@@ -23,7 +23,11 @@ export class SystemTaskHandler implements TaskHandler {
 		private readonly onRunError: (error: unknown) => void,
 	) {}
 
-	async execute(task: ClaimedTask, report: DispatchReporter): Promise<DispatchDecision> {
+	async execute(
+		task: ClaimedTask,
+		report: DispatchReporter,
+		leaseSignal: AbortSignal,
+	): Promise<DispatchDecision> {
 		const decision =
 			this.systemTask.effects === 'non-idempotent' ? report.dispatched() : report.notDispatched();
 
@@ -33,6 +37,7 @@ export class SystemTaskHandler implements TaskHandler {
 			this.systemTask,
 			'durable',
 			this.shutdownSignal,
+			leaseSignal,
 		);
 		if (outcome.rejected) {
 			// An aborted run is not reported, but its rejection still propagates so
@@ -43,6 +48,10 @@ export class SystemTaskHandler implements TaskHandler {
 			throw outcome.error;
 		}
 
+		// A clean stop after lease loss must reject so the executor counts the failed attempt.
+		// A clean stop on shutdown still completes.
+		leaseSignal.throwIfAborted();
+
 		this.logger.debug('Ran a system task occurrence', {
 			name: this.systemTask.name,
 			taskId: task.id,

packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts

@@ -25,14 +25,15 @@ async function startRun(task: Pick<SystemTask, 'run'>, signal: AbortSignal): Pro
 
 /** How one observed run ended. A rejected run carries its error. */
 export type SystemTaskRunOutcome =
-	| { result: 'success' | 'aborted'; rejected: false }
-	| { result: 'failure' | 'aborted'; rejected: true; error: unknown };
+	| { result: 'success' | 'aborted' | 'lease_lost'; rejected: false }
+	| { result: 'failure' | 'aborted' | 'lease_lost'; rejected: true; error: unknown };
 
 /**
  * Run one occurrence of `task` with `signal`, emit its start and its settlement,
- * paired by construction, and record one `system_task.run` span around it. Never
- * rejects: a run that settles after `signal` aborted ends as `aborted`, otherwise
- * a resolution ends as `success` and a rejection as `failure`.
+ * paired by construction, and record one `system_task.run` span around it. The
+ * run also stops when `leaseSignal` aborts. Never rejects: a run that settles
+ * after `leaseSignal` aborted ends as `lease_lost`, after `signal` aborted as
+ * `aborted`, otherwise a resolution ends as `success` and a rejection as `failure`.
  */
 export async function observeSystemTaskRun(
 	eventService: EventService,
@@ -40,8 +41,16 @@ export async function observeSystemTaskRun(
 	task: Pick<SystemTask, 'name' | 'run'>,
 	mode: SystemTaskMode,
 	signal: AbortSignal,
+	leaseSignal?: AbortSignal,
 ): Promise<SystemTaskRunOutcome> {
 	const { name } = task;
+	const runSignal = leaseSignal ? AbortSignal.any([signal, leaseSignal]) : signal;
+	const abortResult = (): 'lease_lost' | 'aborted' | undefined => {
+		if (leaseSignal?.aborted) {
+			return 'lease_lost';
+		}
+		return signal.aborted ? 'aborted' : undefined;
+	};
 	const spanOptions = {
 		name: 'System task run',
 		op: 'system_task.run',
@@ -59,13 +68,13 @@ export async function observeSystemTaskRun(
 	return await startSpan(spanOptions, async (span: Span) => {
 		const startedAt = performance.now();
 		emitSystemTaskMetric(eventService, 'system-task-run-started', { name, mode });
-		const outcome = await startRun(task, signal).then(
+		const outcome = await startRun(task, runSignal).then(
 			(): SystemTaskRunOutcome => ({
-				result: signal.aborted ? 'aborted' : 'success',
+				result: abortResult() ?? 'success',
 				rejected: false,
 			}),
 			(error: unknown): SystemTaskRunOutcome => ({
-				result: signal.aborted ? 'aborted' : 'failure',
+				result: abortResult() ?? 'failure',
 				rejected: true,
 				error,
 			}),
@@ -77,7 +86,7 @@ export async function observeSystemTaskRun(
 			durationMs: performance.now() - startedAt,
 		});
 		span.setAttribute(SYSTEM_TASK_ATTRIBUTES.result, outcome.result);
-		// An abort is an expected shutdown, so only a real failure errors the span.
+		// A shutdown or a lost lease is an expected stop, so only a real failure errors the span.
 		if (outcome.result === 'failure') {
 			span.setStatus({ code: SpanStatus.error, message: ensureError(outcome.error).message });
 		} else {