From e82087bcc8f8c2e8574a4f42b9019b06329be76d Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Tue, 6 Oct 2026 09:07:37 +0000 Subject: [PATCH] feat(core): Stop a durable system task run when its claim is lost (no-changelog) (#40272) --- .../decorators/src/system-task/system-task.ts | 10 +- packages/@n8n/scheduler/README.md | 6 +- .../maps/system-task-metrics.event-map.ts | 5 +- .../prometheus/system-task-metrics.service.ts | 2 +- .../__tests__/system-task-handler.test.ts | 110 ++++++++++++++---- .../system-task-run-observer.test.ts | 27 +++++ .../system-tasks/system-task-handler.ts | 15 ++- .../system-tasks/system-task-run-observer.ts | 27 +++-- 8 files changed, 158 insertions(+), 44 deletions(-) diff --git a/packages/@n8n/decorators/src/system-task/system-task.ts b/packages/@n8n/decorators/src/system-task/system-task.ts index e310dd49c2c..320137b63ae 100644 --- a/packages/@n8n/decorators/src/system-task/system-task.ts +++ b/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; } diff --git a/packages/@n8n/scheduler/README.md b/packages/@n8n/scheduler/README.md index 60a52fa7639..7615ac3b484 100644 --- a/packages/@n8n/scheduler/README.md +++ b/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 diff --git a/packages/cli/src/events/maps/system-task-metrics.event-map.ts b/packages/cli/src/events/maps/system-task-metrics.event-map.ts index 56f2d23b36a..f3e44549780 100644 --- a/packages/cli/src/events/maps/system-task-metrics.event-map.ts +++ b/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 diff --git a/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts b/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts index 710efb65485..eb0c7c8802c 100644 --- a/packages/cli/src/metrics/prometheus/system-task-metrics.service.ts +++ b/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, }); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts index 068bef3e993..a1b787cd037 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-handler.test.ts +++ b/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(); 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((_, 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((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((_, 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' }), + ); + }); }); }); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts index 0f0e3067223..fc4fc70deb8 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts +++ b/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(); + 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 }), + ); + }); }); diff --git a/packages/cli/src/scheduling/system-tasks/system-task-handler.ts b/packages/cli/src/scheduling/system-tasks/system-task-handler.ts index 903c43274a0..615ee11c0bd 100644 --- a/packages/cli/src/scheduling/system-tasks/system-task-handler.ts +++ b/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 { + async execute( + task: ClaimedTask, + report: DispatchReporter, + leaseSignal: AbortSignal, + ): Promise { 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, diff --git a/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts b/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts index 1deaed1dacf..aca7bdcc23c 100644 --- a/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts +++ b/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts @@ -25,14 +25,15 @@ async function startRun(task: Pick, 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, mode: SystemTaskMode, signal: AbortSignal, + leaseSignal?: AbortSignal, ): Promise { 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 {