fix(core): Count jobs as running from the start of job processing (#40050)

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

packages/cli/src/scaling/__tests__/job-processor.service.test.ts

@@ -33,7 +33,7 @@ import {
 } from 'n8n-workflow';
 import type PCancelable from 'p-cancelable';
 import type { Mock, MockedClass, MockInstance } from 'vitest';
-import { mock } from 'vitest-mock-extended';
+import { mock, type MockProxy } from 'vitest-mock-extended';
 import { z } from 'zod';
 
 import { CredentialsHelper } from '@/credentials-helper';
@@ -259,7 +259,113 @@ describe('JobProcessor', () => {
 
 		await expect(jobProcessor.processJob(job)).rejects.toThrow('workflow run rejected');
 
-		expect(jobProcessor.getRunningJobIds()).toEqual([]);
+		expect(jobProcessor.getTrackedJobIds()).toEqual([]);
+	});
+
+	describe('jobs in preflight', () => {
+		const createJobProcessor = (executionPersistence: ExecutionPersistence) =>
+			new JobProcessor(
+				logger,
+				mock<ExecutionRepository>(),
+				executionPersistence,
+				mock(),
+				mock(),
+				mock(),
+				createManualExecutionServiceMock(),
+				executionsConfig,
+				mock(),
+				mock(),
+			);
+
+		const createPendingExecution = (executionPersistence: MockProxy<ExecutionPersistence>) => {
+			let resolveExecution: (execution: IExecutionResponse) => void = () => {};
+			executionPersistence.findSingleExecution.mockReturnValue(
+				new Promise<IExecutionResponse>((resolve) => (resolveExecution = resolve)),
+			);
+			return () => resolveExecution(mock<IExecutionResponse>({ status: 'crashed' }));
+		};
+
+		it('should list a job as running and in preflight, but not in the running jobs summary, while its execution is still being loaded', async () => {
+			const executionPersistence = mock<ExecutionPersistence>();
+			const settleExecution = createPendingExecution(executionPersistence);
+			const jobProcessor = createJobProcessor(executionPersistence);
+			const job = mock<Job>({
+				id: 'job-1',
+				data: { executionId: 'exec-1', loadStaticData: false },
+			});
+
+			const processing = jobProcessor.processJob(job);
+
+			expect(jobProcessor.getTrackedJobIds()).toContain('job-1');
+			expect(jobProcessor.getRunningJobsSummary()).toEqual([]);
+			expect(jobProcessor.getJobsInPreflight()).toEqual([
+				{ jobId: 'job-1', executionId: 'exec-1' },
+			]);
+
+			settleExecution();
+			await processing;
+			expect(jobProcessor.getTrackedJobIds()).not.toContain('job-1');
+			expect(jobProcessor.getJobsInPreflight()).toEqual([]);
+		});
+
+		it('should list a job as running but no longer in preflight once its execution has loaded and its run has started', async () => {
+			const executionRepository = mock<ExecutionRepository>();
+			const executionPersistence = mock<ExecutionPersistence>();
+			executionPersistence.findSingleExecution.mockResolvedValueOnce(
+				mock<IExecutionResponse>({
+					mode: 'manual',
+					workflowData: { id: 'workflow-id', nodes: [], staticData: {} },
+					data: mock<IRunExecutionData>({ executionData: undefined }),
+				}),
+			);
+
+			const cancel = vi.fn();
+			const runPromise = new Promise<IRun>(() => {});
+			const workflowRun = Object.assign(runPromise, { cancel }) as unknown as PCancelable<IRun>;
+
+			const manualExecutionService = mock<ManualExecutionService>();
+			manualExecutionService.runManually.mockReturnValue(workflowRun);
+
+			const jobProcessor = new JobProcessor(
+				logger,
+				executionRepository,
+				executionPersistence,
+				mock(),
+				mock(),
+				mock(),
+				manualExecutionService,
+				executionsConfig,
+				mock(),
+				mock(),
+			);
+
+			const job = mock<Job>({
+				id: 'job-1',
+				data: { executionId: 'exec-1', loadStaticData: false },
+			});
+
+			void jobProcessor.processJob(job);
+
+			await vi.waitFor(() => expect(jobProcessor.getRunningJobsSummary()).not.toEqual([]));
+
+			expect(jobProcessor.getJobsInPreflight()).toEqual([]);
+			expect(jobProcessor.getTrackedJobIds()).toContain('job-1');
+		});
+
+		it('should stop tracking a job when loading its execution fails', async () => {
+			const executionPersistence = mock<ExecutionPersistence>();
+			executionPersistence.findSingleExecution.mockRejectedValue(new Error('database unavailable'));
+			const jobProcessor = createJobProcessor(executionPersistence);
+			const job = mock<Job>({
+				id: 'job-1',
+				data: { executionId: 'exec-1', loadStaticData: false },
+			});
+
+			await expect(jobProcessor.processJob(job)).rejects.toThrow('database unavailable');
+
+			expect(jobProcessor.getTrackedJobIds()).not.toContain('job-1');
+			expect(jobProcessor.getJobsInPreflight()).toEqual([]);
+		});
 	});
 
 	it('should send job-finished with success=false when execution has errors', async () => {
@@ -2522,13 +2628,13 @@ describe('JobProcessor', () => {
 
 			// Flush the microtasks processJob runs before it starts tracking the job.
 			await vi.advanceTimersByTimeAsync(0);
-			expect(jobProcessor.getRunningJobIds()).toContain(String(job.id));
+			expect(jobProcessor.getTrackedJobIds()).toContain(String(job.id));
 
 			rejectRun(new Error('worker crashed'));
 			await assertion;
 
 			// The running-job entry must be gone, not left dangling as still "running".
-			expect(jobProcessor.getRunningJobIds()).not.toContain(String(job.id));
+			expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id));
 
 			// The watchdog must be cleared too: advancing past the workflow timeout must not
 			// trigger a delayed, spurious cancellation of an already-failed execution.
@@ -2599,6 +2705,56 @@ describe('JobProcessor', () => {
 			);
 		});
 
+		it('should stop counting a job as running once it is cancelled, before its run settles', async () => {
+			vi.useFakeTimers();
+
+			const executionRepository = mock<ExecutionRepository>();
+			const executionPersistence = mock<ExecutionPersistence>();
+			executionPersistence.findSingleExecution.mockResolvedValueOnce(
+				mock<IExecutionResponse>({
+					mode: 'manual',
+					workflowData: {
+						id: 'workflow-id',
+						nodes: [],
+						staticData: {},
+						settings: { executionTimeout: 5 },
+					},
+					data: mock<IRunExecutionData>({ executionData: undefined }),
+				}),
+			);
+
+			const cancel = vi.fn();
+			const runPromise = new Promise<IRun>(() => {});
+			const workflowRun = Object.assign(runPromise, { cancel }) as unknown as PCancelable<IRun>;
+
+			const manualExecutionService = mock<ManualExecutionService>();
+			manualExecutionService.runManually.mockReturnValue(workflowRun);
+
+			const jobProcessor = new JobProcessor(
+				logger,
+				executionRepository,
+				executionPersistence,
+				mock(),
+				mock(),
+				mock(),
+				manualExecutionService,
+				executionsConfig,
+				mock(),
+				mock(),
+			);
+
+			const job = mock<Job>({ data: { executionId: 'execution-id', loadStaticData: false } });
+
+			void jobProcessor.processJob(job);
+
+			await vi.advanceTimersByTimeAsync(0);
+			expect(jobProcessor.getTrackedJobIds()).toContain(String(job.id));
+
+			jobProcessor.stopJob(job.id);
+
+			expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id));
+		});
+
 		it('keeps a stop reported as manual when the timeout elapses before the run settles', async () => {
 			vi.useFakeTimers();
 
@@ -2714,7 +2870,7 @@ describe('JobProcessor', () => {
 			await vi.advanceTimersByTimeAsync(10_000);
 
 			expect(cancel).not.toHaveBeenCalled();
-			expect(jobProcessor.getRunningJobIds()).not.toContain(String(job.id));
+			expect(jobProcessor.getTrackedJobIds()).not.toContain(String(job.id));
 			expect(eventService.emit).not.toHaveBeenCalledWith('execution-cancelled', expect.anything());
 		});
 

packages/cli/src/scaling/__tests__/scaling.service.test.ts

@@ -135,6 +135,7 @@ describe('ScalingService', () => {
 		activeExecutions.getRunningExecutionIds.mockReturnValue([]);
 		activeExecutions.cancelRunningExecutions.mockResolvedValue([]);
 		jobProcessor.getRunningJobsSummary.mockReturnValue([]);
+		jobProcessor.getJobsInPreflight.mockReturnValue([]);
 		globalConfig.generic.gracefulShutdownTimeout = 30;
 
 		scalingService = new ScalingService(
@@ -336,6 +337,40 @@ describe('ScalingService', () => {
 			);
 			expect(errorReporter.error).toHaveBeenCalledWith(originalError, { executionId: '123' });
 		});
+
+		it('should warn once when a job reaches the worker after shutdown began', async () => {
+			// @ts-expect-error readonly property
+			instanceSettings.instanceType = 'worker';
+			await scalingService.setupQueue();
+			scalingService.setupWorker(5);
+			const processFn = queue.process.mock.calls[0][2] as unknown as (job: Job) => Promise<void>;
+			jobProcessor.getTrackedJobIds.mockReturnValue([]);
+
+			await scalingService.stop();
+
+			const job = mock<Job>({ id: '1', data: { executionId: '123', loadStaticData: false } });
+			await processFn(job);
+
+			expect(scopedLogger.warn).toHaveBeenCalledTimes(1);
+			expect(scopedLogger.warn).toHaveBeenCalledWith(
+				expect.stringContaining('123'),
+				expect.objectContaining({ executionId: '123', jobId: '1' }),
+			);
+		});
+
+		it('should process a job that reaches the worker before shutdown without warning', async () => {
+			// @ts-expect-error readonly property
+			instanceSettings.instanceType = 'worker';
+			await scalingService.setupQueue();
+			scalingService.setupWorker(5);
+			const processFn = queue.process.mock.calls[0][2] as unknown as (job: Job) => Promise<void>;
+
+			const job = mock<Job>({ id: '1', data: { executionId: '123', loadStaticData: false } });
+			await processFn(job);
+
+			expect(scopedLogger.warn).not.toHaveBeenCalled();
+			expect(jobProcessor.processJob).toHaveBeenCalledWith(job);
+		});
 	});
 
 	describe('stop', () => {
@@ -373,7 +408,7 @@ describe('ScalingService', () => {
 				// @ts-expect-error readonly property
 				instanceSettings.instanceType = 'worker';
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 
 				await scalingService.stop();
 
@@ -387,7 +422,7 @@ describe('ScalingService', () => {
 				// @ts-expect-error readonly property
 				instanceSettings.instanceType = 'worker';
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValueOnce(['1']).mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValueOnce(['1']).mockReturnValue([]);
 				jobProcessor.getRunningJobsSummary.mockReturnValue([mock({ executionId: 'exec-1' })]);
 
 				const stopped = scalingService.stop();
@@ -400,12 +435,30 @@ describe('ScalingService', () => {
 				);
 			});
 
+			it('should log the execution IDs of jobs still in preflight while draining', async () => {
+				vi.useFakeTimers();
+				// @ts-expect-error readonly property
+				instanceSettings.instanceType = 'worker';
+				await scalingService.setupQueue();
+				jobProcessor.getTrackedJobIds.mockReturnValueOnce(['1']).mockReturnValue([]);
+				jobProcessor.getJobsInPreflight.mockReturnValue([{ jobId: '1', executionId: 'exec-1' }]);
+
+				const stopped = scalingService.stop();
+				await vi.advanceTimersByTimeAsync(500);
+				await stopped;
+
+				expect(scopedLogger.info).toHaveBeenCalledWith(
+					'Waiting for 1 executions to start... (execution IDs: exec-1)',
+					{ executionIds: ['exec-1'] },
+				);
+			});
+
 			it('should keep waiting for an in-process execution that has no queue job', async () => {
 				vi.useFakeTimers();
 				// @ts-expect-error readonly property
 				instanceSettings.instanceType = 'worker';
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 
 				let inProcessExecutionIds = ['exec-1'];
 				activeExecutions.getRunningExecutionIds.mockImplementation(() => inProcessExecutionIds);
@@ -436,7 +489,7 @@ describe('ScalingService', () => {
 				// The budget is 80% of the shutdown window, so 4s of the 5s here.
 				globalConfig.generic.gracefulShutdownTimeout = 5;
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 				activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']);
 				activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']);
 
@@ -468,7 +521,7 @@ describe('ScalingService', () => {
 				instanceSettings.instanceType = 'worker';
 				globalConfig.generic.gracefulShutdownTimeout = 5;
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 				activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']);
 
 				let finishCancellation: (executionIds: string[]) => void = () => {};
@@ -500,7 +553,7 @@ describe('ScalingService', () => {
 				instanceSettings.instanceType = 'worker';
 				globalConfig.generic.gracefulShutdownTimeout = 1;
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 
 				let inProcessExecutionIds = ['exec-1'];
 				activeExecutions.getRunningExecutionIds.mockImplementation(() => inProcessExecutionIds);
@@ -535,7 +588,7 @@ describe('ScalingService', () => {
 					await scalingService.setupQueue();
 
 					let runningJobIds = ['1'];
-					jobProcessor.getRunningJobIds.mockImplementation(() => runningJobIds);
+					jobProcessor.getTrackedJobIds.mockImplementation(() => runningJobIds);
 					activeExecutions.getRunningExecutionIds.mockReturnValue(inProcessExecutionIds);
 					activeExecutions.cancelRunningExecutions.mockResolvedValue(inProcessExecutionIds);
 
@@ -567,7 +620,7 @@ describe('ScalingService', () => {
 				// The budget is 800ms, which the force-exit timer at 1s must not beat.
 				globalConfig.generic.gracefulShutdownTimeout = 1;
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 				activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']);
 				activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']);
 
@@ -588,7 +641,7 @@ describe('ScalingService', () => {
 				instanceSettings.instanceType = 'worker';
 				globalConfig.generic.gracefulShutdownTimeout = 0;
 				await scalingService.setupQueue();
-				jobProcessor.getRunningJobIds.mockReturnValue([]);
+				jobProcessor.getTrackedJobIds.mockReturnValue([]);
 				activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']);
 
 				let hasStopped = false;
@@ -618,7 +671,7 @@ describe('ScalingService', () => {
 					instanceSettings.instanceType = 'worker';
 					globalConfig.generic.gracefulShutdownTimeout = shutdownTimeout;
 					await scalingService.setupQueue();
-					jobProcessor.getRunningJobIds.mockReturnValue([]);
+					jobProcessor.getTrackedJobIds.mockReturnValue([]);
 					activeExecutions.getRunningExecutionIds.mockReturnValue(['exec-1']);
 					activeExecutions.cancelRunningExecutions.mockResolvedValue(['exec-1']);
 
@@ -1010,7 +1063,7 @@ describe('ScalingService', () => {
 			await scalingService.setupQueue();
 			queue.eventNames.mockReturnValue(['global:progress', 'global:completed']);
 			queue.listenerCount.mockImplementation((event) => (event === 'global:completed' ? 2 : 1));
-			jobProcessor.getRunningJobIds.mockReturnValue(['job-1']);
+			jobProcessor.getTrackedJobIds.mockReturnValue(['job-1']);
 
 			const messageHandler = queue.on.mock.calls.find(
 				([event]) => (event as string) === 'global:progress',

packages/cli/src/scaling/job-processor.ts

@@ -100,6 +100,9 @@ function scheduleAt(timestamp: number, fn: () => void): () => void {
 export class JobProcessor {
 	private readonly runningJobs: Record<JobId, RunningJob> = {};
 
+	/** Execution id per job id for every job in `processJob`, including jobs still in preflight. */
+	private readonly trackedJobs = new Map<string, string>();
+
 	/** Cause of the cancellation of each job cancelled so far, kept until its run settles. */
 	private readonly cancellationReasons: Record<JobId, CancellationReason> = {};
 
@@ -119,6 +122,15 @@ export class JobProcessor {
 	}
 
 	async processJob(job: Job): Promise<JobResult> {
+		this.trackedJobs.set(String(job.id), job.data.executionId);
+		try {
+			return await this.runJob(job);
+		} finally {
+			this.trackedJobs.delete(String(job.id));
+		}
+	}
+
+	private async runJob(job: Job): Promise<JobResult> {
 		const { executionId, loadStaticData } = job.data;
 
 		const execution = await this.executionPersistence.findSingleExecution(executionId, {
@@ -580,11 +592,20 @@ export class JobProcessor {
 
 		runningJob.run.cancel();
 		delete this.runningJobs[jobId];
+		// The run may ignore cancellation and never settle; drop tracking now instead of waiting for it.
+		this.trackedJobs.delete(String(jobId));
 		this.cancellationReasons[jobId] = reason;
 	}
 
-	getRunningJobIds(): JobId[] {
-		return Object.keys(this.runningJobs);
+	/** Ids of the jobs tracked from the start of processing, including jobs still in preflight. */
+	getTrackedJobIds(): JobId[] {
+		return [...this.trackedJobs.keys()];
+	}
+
+	getJobsInPreflight(): Array<{ jobId: JobId; executionId: string }> {
+		return [...this.trackedJobs]
+			.filter(([jobId]) => !(jobId in this.runningJobs))
+			.map(([jobId, executionId]) => ({ jobId, executionId }));
 	}
 
 	getRunningJobsSummary(): RunningJobSummary[] {

packages/cli/src/scaling/scaling.service.ts

@@ -53,6 +53,8 @@ export class ScalingService {
 
 	private createBullQueue?: (name: string) => JobQueue;
 
+	private stopping = false;
+
 	constructor(
 		private readonly logger: Logger,
 		private readonly errorReporter: ErrorReporter,
@@ -169,6 +171,16 @@ export class ScalingService {
 		this.assertQueue();
 
 		void this.defaultQueue.process(JOB_TYPE_NAME, concurrency, async (job: Job) => {
+			// The job still runs: the handler passes it to JobProcessor, which tracks it and holds the drain.
+			if (this.stopping) {
+				const { executionId } = job.data;
+				const jobId = job.id;
+				this.logger.warn(
+					`Worker received job ${jobId} for execution ${executionId} after it began to stop`,
+					{ executionId, jobId },
+				);
+			}
+
 			try {
 				this.eventService.emit('job-dequeued', {
 					executionId: job.data.executionId,
@@ -249,6 +261,8 @@ export class ScalingService {
 	}
 
 	private async stopWorker() {
+		this.stopping = true;
+
 		await this.pauseAllQueues();
 
 		const shutdownWindowMs =
@@ -321,6 +335,17 @@ export class ScalingService {
 			);
 		}
 
+		const preflightExecutionIds = this.jobProcessor
+			.getJobsInPreflight()
+			.map(({ executionId }) => executionId);
+
+		if (preflightExecutionIds.length > 0) {
+			this.logger.info(
+				`Waiting for ${preflightExecutionIds.length} executions to start... (execution IDs: ${preflightExecutionIds.join(', ')})`,
+				{ executionIds: preflightExecutionIds },
+			);
+		}
+
 		const inProcessExecutionIds = this.activeExecutions.getRunningExecutionIds();
 
 		if (inProcessExecutionIds.length > 0) {
@@ -445,7 +470,7 @@ export class ScalingService {
 	}
 
 	getRunningJobsCount() {
-		return this.jobProcessor.getRunningJobIds().length;
+		return this.jobProcessor.getTrackedJobIds().length;
 	}
 
 	/**