From 839b841dda2a10b2caeed1afe5bfb63428a044f6 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Mon, 12 Oct 2026 10:46:11 +0000 Subject: [PATCH] feat(core): Stop a scheduled task run at its timeout (no-changelog) (#40515) Co-authored-by: Veljko Terzic --- packages/@n8n/scheduler/README.md | 7 +- .../src/core/__tests__/factory.test.ts | 30 +-- packages/@n8n/scheduler/src/core/errors.ts | 11 + .../core/executor/__tests__/executor.test.ts | 214 +++++++++++++----- .../scheduler/src/core/executor/executor.ts | 69 +++--- .../@n8n/scheduler/src/core/executor/index.ts | 1 + .../src/core/executor/lease-constants.ts | 9 +- .../src/core/executor/task-handler.ts | 23 +- packages/@n8n/scheduler/src/core/factory.ts | 7 +- packages/@n8n/scheduler/src/core/index.ts | 2 + .../__tests__/executor-tracing.test.ts | 17 +- .../src/observability/executor-tracing.ts | 6 +- .../scheduler/src/observability/metrics.ts | 3 + .../scheduler-metrics.service.test.ts | 9 + .../prometheus/scheduler-metrics.service.ts | 13 ++ .../poll-trigger-task-handler.test.ts | 107 +++++---- .../poll-trigger-task-handler.ts | 10 +- .../__tests__/system-task-handler.test.ts | 48 ++-- .../__tests__/system-task-runner.test.ts | 10 +- .../system-tasks/system-task-handler.ts | 10 +- .../scheduling/scheduler-task-timeout.test.ts | 104 +++++++++ 21 files changed, 500 insertions(+), 210 deletions(-) create mode 100644 packages/cli/test/integration/scheduling/scheduler-task-timeout.test.ts diff --git a/packages/@n8n/scheduler/README.md b/packages/@n8n/scheduler/README.md index 7615ac3b484..5c76a1ab4a1 100644 --- a/packages/@n8n/scheduler/README.md +++ b/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 diff --git a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts b/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts index 67965ba352b..3bb3729d71c 100644 --- a/packages/@n8n/scheduler/src/core/__tests__/factory.test.ts +++ b/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()); + it('counts and warns once when a run reaches its timeout', async () => { + const metrics = mock(); + 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(); }); }); diff --git a/packages/@n8n/scheduler/src/core/errors.ts b/packages/@n8n/scheduler/src/core/errors.ts index 566120683b3..798bcbd3de7 100644 --- a/packages/@n8n/scheduler/src/core/errors.ts +++ b/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) { diff --git a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts b/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts index 364b507566f..9deead2ffb6 100644 --- a/packages/@n8n/scheduler/src/core/executor/__tests__/executor.test.ts +++ b/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) => { 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((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((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(() => {}); + }), + }); + 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(() => {}); + }), + }); + 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 () => { diff --git a/packages/@n8n/scheduler/src/core/executor/executor.ts b/packages/@n8n/scheduler/src/core/executor/executor.ts index fee41f199e5..61c4c2114a4 100644 --- a/packages/@n8n/scheduler/src/core/executor/executor.ts +++ b/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 { 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 { - 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(); } } diff --git a/packages/@n8n/scheduler/src/core/executor/index.ts b/packages/@n8n/scheduler/src/core/executor/index.ts index 8347dfa6f2d..242dba42370 100644 --- a/packages/@n8n/scheduler/src/core/executor/index.ts +++ b/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'; diff --git a/packages/@n8n/scheduler/src/core/executor/lease-constants.ts b/packages/@n8n/scheduler/src/core/executor/lease-constants.ts index 0c4f768ec48..a53451fb03d 100644 --- a/packages/@n8n/scheduler/src/core/executor/lease-constants.ts +++ b/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; diff --git a/packages/@n8n/scheduler/src/core/executor/task-handler.ts b/packages/@n8n/scheduler/src/core/executor/task-handler.ts index 76053f69b25..7d96415bc87 100644 --- a/packages/@n8n/scheduler/src/core/executor/task-handler.ts +++ b/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; + execute(task: ClaimedTask, report: DispatchReporter, run: TaskRun): Promise; +} + +/** 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; } /** diff --git a/packages/@n8n/scheduler/src/core/factory.ts b/packages/@n8n/scheduler/src/core/factory.ts index aa0486ca284..76408a3f39e 100644 --- a/packages/@n8n/scheduler/src/core/factory.ts +++ b/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) => { diff --git a/packages/@n8n/scheduler/src/core/index.ts b/packages/@n8n/scheduler/src/core/index.ts index cdffce9ff43..373dfcabf69 100644 --- a/packages/@n8n/scheduler/src/core/index.ts +++ b/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'; diff --git a/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts b/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts index 2f33eb4a3d4..877f4ba65f9 100644 --- a/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts +++ b/packages/@n8n/scheduler/src/observability/__tests__/executor-tracing.test.ts @@ -151,12 +151,12 @@ describe('withHandoffTracing', () => { }; const task = claimedTask(); const report = mock(); - 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(), - new AbortController().signal, - ), + withHandoffTracing(tracer, handler).execute(claimedTask(), mock(), { + signal: new AbortController().signal, + remainingMs: () => 45_000, + }), ).rejects.toThrow('boom'); expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatus.error, message: 'boom' }); diff --git a/packages/@n8n/scheduler/src/observability/executor-tracing.ts b/packages/@n8n/scheduler/src/observability/executor-tracing.ts index c40e0aa0306..4aafa0d631f 100644 --- a/packages/@n8n/scheduler/src/observability/executor-tracing.ts +++ b/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; }, diff --git a/packages/@n8n/scheduler/src/observability/metrics.ts b/packages/@n8n/scheduler/src/observability/metrics.ts index 5fab4be2f33..3b3ac6da37b 100644 --- a/packages/@n8n/scheduler/src/observability/metrics.ts +++ b/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() {}, diff --git a/packages/cli/src/metrics/prometheus/__tests__/scheduler-metrics.service.test.ts b/packages/cli/src/metrics/prometheus/__tests__/scheduler-metrics.service.test.ts index e6dcc826c31..4edd071306c 100644 --- a/packages/cli/src/metrics/prometheus/__tests__/scheduler-metrics.service.test.ts +++ b/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(); diff --git a/packages/cli/src/metrics/prometheus/scheduler-metrics.service.ts b/packages/cli/src/metrics/prometheus/scheduler-metrics.service.ts index c1a17346df5..d827bf246e8 100644 --- a/packages/cli/src/metrics/prometheus/scheduler-metrics.service.ts +++ b/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); diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts index 8602d3b3e09..5ac5186a053 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts +++ b/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(); @@ -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]; diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts index 3eba1095434..39cce2c8822 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts +++ b/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 { // 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. 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 1b1edc9890a..8eb219f1024 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,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({ 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(); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts index 85b44e00cc6..14681b48f6b 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts +++ b/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 { await runner.init(); } @@ -790,7 +792,7 @@ describe('SystemTaskRunner', () => { handler.execute( mock(), createDispatchReporter(vi.fn()), - new AbortController().signal, + runOf(new AbortController().signal), ), ).rejects.toThrow(error); @@ -819,7 +821,7 @@ describe('SystemTaskRunner', () => { const executing = handler.execute( mock(), 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(), createDispatchReporter(vi.fn()), - new AbortController().signal, + runOf(new AbortController().signal), ); expect(eventService.emit).toHaveBeenCalledWith('system-task-run-started', { 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 5ecc91de4d8..e5ce535e800 100644 --- a/packages/cli/src/scheduling/system-tasks/system-task-handler.ts +++ b/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 { const outcome = await observeSystemTaskRun( this.eventService, diff --git a/packages/cli/test/integration/scheduling/scheduler-task-timeout.test.ts b/packages/cli/test/integration/scheduling/scheduler-task-timeout.test.ts new file mode 100644 index 00000000000..8fed984d242 --- /dev/null +++ b/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(); + 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(() => {}); + }, + }); + }); + + 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); +});