fix(core): Stop waiting on Bull job.finished() for queued executions (#39723)

packages/cli/src/__tests__/workflow-runner.test.ts

@@ -33,6 +33,7 @@ import {
 	type WorkflowExecuteMode,
 	Workflow,
 	ExecutionError,
+	WorkflowOperationError,
 	TimeoutExecutionCancelledError,
 	createRunExecutionData,
 } from 'n8n-workflow';
@@ -56,6 +57,7 @@ import { OwnershipService } from '@/services/ownership.service';
 import { Telemetry } from '@/telemetry';
 import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
 import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
+import type { Job } from '@/scaling/scaling.types';
 import { WorkflowRunner } from '@/workflow-runner';
 
 // `@/scaling/scaling.service` is dynamically imported by `enqueueExecution`.
@@ -64,12 +66,18 @@ import { WorkflowRunner } from '@/workflow-runner';
 // class isn't initialised when the hoisted factory first resolves.
 const setupQueue = vi.fn();
 const addJob = vi.fn();
+const waitForJob = vi.fn();
+const popJobResult = vi.fn();
 
 @Service()
 class MockScalingService {
 	setupQueue = setupQueue;
 
 	addJob = addJob;
+
+	waitForJob = waitForJob;
+
+	popJobResult = popJobResult;
 }
 
 vi.mock('@/scaling/scaling.service', () => ({
@@ -1085,6 +1093,70 @@ describe('enqueueExecution', () => {
 		expect(setupQueue).toHaveBeenCalledTimes(1);
 	});
 
+	it('should fail the execution when the result cannot be read from the DB after the job ended', async () => {
+		const activeExecutions = Container.get(ActiveExecutions);
+		let workflowExecution: PCancelable<IRun> | undefined;
+		vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockImplementation((_, execution) => {
+			workflowExecution = execution;
+		});
+		const processError = vi.spyOn(runner, 'processError').mockResolvedValue();
+		const data = mock<IWorkflowExecutionDataProcess>({
+			workflowData: { nodes: [], staticData: {} },
+			executionData: undefined,
+		});
+		const readError = new Error('db unavailable');
+		addJob.mockResolvedValueOnce(mock<Job>({ id: 'job-1', data: { executionId: '1' } }));
+		waitForJob.mockResolvedValueOnce(undefined);
+		popJobResult.mockReturnValueOnce(undefined);
+		vi.spyOn(Container.get(ExecutionPersistence), 'findSingleExecution').mockRejectedValueOnce(
+			readError,
+		);
+
+		// @ts-expect-error Private method
+		await runner.enqueueExecution('1', 'workflow-xyz', data);
+
+		await expect(workflowExecution).rejects.toThrowError(readError);
+		// A failed run makes the webhook answer with an error instead of a success without data
+		expect(processError).toHaveBeenCalledWith(
+			readError,
+			expect.any(Date),
+			data.executionMode,
+			'1',
+			expect.anything(),
+		);
+	});
+
+	it('should fail the execution when its record is gone after the job ended', async () => {
+		const activeExecutions = Container.get(ActiveExecutions);
+		let workflowExecution: PCancelable<IRun> | undefined;
+		vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockImplementation((_, execution) => {
+			workflowExecution = execution;
+		});
+		const processError = vi.spyOn(runner, 'processError').mockResolvedValue();
+		const data = mock<IWorkflowExecutionDataProcess>({
+			workflowData: { nodes: [], staticData: {} },
+			executionData: undefined,
+		});
+		addJob.mockResolvedValueOnce(mock<Job>({ id: 'job-1', data: { executionId: '1' } }));
+		waitForJob.mockResolvedValueOnce(undefined);
+		popJobResult.mockReturnValueOnce(undefined);
+		vi.spyOn(Container.get(ExecutionPersistence), 'findSingleExecution').mockResolvedValueOnce(
+			undefined,
+		);
+
+		// @ts-expect-error Private method
+		await runner.enqueueExecution('1', 'workflow-xyz', data);
+
+		// An unsaved execution is deleted by the worker, so a missing record is not a bug
+		await expect(workflowExecution).rejects.toThrowError('Could not find execution with id "1"');
+		expect(processError).toHaveBeenCalledWith(
+			expect.any(WorkflowOperationError),
+			expect.any(Date),
+			data.executionMode,
+			'1',
+			expect.anything(),
+		);
+	});
 	it('should finalize the execution when pool resolution fails', async () => {
 		const activeExecutions = Container.get(ActiveExecutions);
 		vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValue();

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

@@ -1,3 +1,5 @@
+import type { ExecutionStatus } from 'n8n-workflow';
+
 export type QueueMetricsEventMap = {
 	'job-counts-updated': {
 		active: number;
@@ -5,4 +7,10 @@ export type QueueMetricsEventMap = {
 		failed: number;
 		waiting: number;
 	};
+
+	/** A main missed every completion event for a queued job and settled it from the DB. */
+	'job-completion-missed': {
+		/** Execution status found in the DB, or `deleted` when its row was gone. */
+		status: ExecutionStatus | 'deleted';
+	};
 };

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

@@ -131,6 +131,30 @@ describe('PrometheusQueueMetricsService', () => {
 		});
 	});
 
+	describe('job-completion-missed event handler', () => {
+		it('should create the completion-missed counter with a status label', () => {
+			service.init();
+
+			expect(promClient.Counter).toHaveBeenCalledWith(
+				expect.objectContaining({
+					name: 'n8n_scaling_mode_queue_jobs_completion_missed',
+					labelNames: ['status'],
+				}),
+			);
+		});
+
+		it('should count a missed completion under its execution status', () => {
+			service.init();
+			const handler = getEventHandler('job-completion-missed');
+			vi.clearAllMocks();
+
+			expect(handler).toBeDefined();
+			handler!({ status: 'deleted' });
+
+			expect(mockCounterInc).toHaveBeenCalledWith({ status: 'deleted' }, 1);
+		});
+	});
+
 	describe('job-counts-updated event handler', () => {
 		it('should update gauges and counters with correct values from job counts', () => {
 			service.init();

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

@@ -59,5 +59,15 @@ export class PrometheusQueueMetricsService implements PrometheusMetricsCollector
 			completedCounter.inc(jobCounts.completed);
 			failedCounter.inc(jobCounts.failed);
 		});
+
+		const completionMissedCounter = new promClient.Counter({
+			name: `${prefix}scaling_mode_queue_jobs_completion_missed`,
+			help: 'Total number of jobs whose completion event this main missed and settled from the database instead, by execution status found there.',
+			labelNames: ['status'],
+		});
+
+		this.eventService.on('job-completion-missed', ({ status }) => {
+			completionMissedCounter.inc({ status }, 1);
+		});
 	}
 }

packages/cli/src/node-execution/execute-node.service.ts

@@ -302,8 +302,8 @@ export class ExecuteNodeService {
 		}
 	}
 
-	/** The post-execute promise does not reliably settle on multi-main (Bull's
-	 *  `job.finished()` behind it), so poll the execution row there — same as
+	/** The post-execute promise does not reliably settle on multi-main
+	 *  (`ScalingService.waitForJob()` behind it), so poll the execution row there — same as
 	 *  chat-hub's `waitForExecutionCompletion`. */
 	private async waitForSettled(executionId: string, signal: AbortSignal): Promise<void> {
 		if (!this.instanceSettings.isMultiMain) {

packages/cli/src/scaling/__tests__/job-outcome-tracker.test.ts

@@ -0,0 +1,306 @@
+import { mockLogger } from '@n8n/backend-test-utils';
+import type { EventService } from '@n8n/backend-services';
+import type { ExecutionRepository } from '@n8n/db';
+import type { ExecutionStatus } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import type { ActiveExecutions } from '@/active-executions';
+
+import { JOB_WAIT_RECHECK_INTERVAL_MS } from '../constants';
+import { JobOutcomeTracker } from '../job-outcome-tracker';
+import type { Job, JobFinishedProps } from '../scaling.types';
+
+describe('JobOutcomeTracker', () => {
+	const activeExecutions = mock<ActiveExecutions>();
+	const executionRepository = mock<ExecutionRepository>();
+	const eventService = mock<EventService>();
+	let tracker: JobOutcomeTracker;
+
+	const job = mock<Job>({ id: 'job-1', data: { executionId: 'exec-1' }, queue: { name: 'jobs' } });
+	const result = mock<JobFinishedProps>({ success: true, status: 'success' });
+
+	const statusRows = (statuses: Record<string, ExecutionStatus>) =>
+		Object.entries(statuses).map(([id, status]) => ({ id, status }));
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		activeExecutions.has.mockReturnValue(true);
+		tracker = new JobOutcomeTracker(
+			mockLogger(),
+			activeExecutions,
+			executionRepository,
+			eventService,
+		);
+	});
+
+	afterEach(() => {
+		tracker.clear();
+		vi.useRealTimers();
+	});
+
+	describe('recordFinished', () => {
+		it('should keep the result for an execution this process enqueued', () => {
+			tracker.recordFinished('exec-1', result);
+
+			expect(tracker.popResult('exec-1')).toBe(result);
+			expect(tracker.popResult('exec-1')).toBeUndefined();
+		});
+
+		it('should ignore the result for an execution this process did not enqueue', () => {
+			// Bull broadcasts progress messages to every main and webhook process
+			activeExecutions.has.mockReturnValue(false);
+
+			tracker.recordFinished('exec-other-main', result);
+
+			expect(tracker.popResult('exec-other-main')).toBeUndefined();
+		});
+	});
+
+	describe('waitFor', () => {
+		it('should resolve when the worker reports the job as finished', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.recordFinished('exec-1', result);
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(tracker.popResult('exec-1')).toBe(result);
+		});
+
+		it('should resolve when an older worker reports the job as finished without a result', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.recordFinished('exec-1');
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(tracker.popResult('exec-1')).toBeUndefined();
+		});
+
+		it('should resolve at once when the result arrived before the wait started', async () => {
+			tracker.recordFinished('exec-1', result);
+
+			await expect(tracker.waitFor(job)).resolves.toBeUndefined();
+		});
+
+		it('should reject when the worker reports the job as failed', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.recordFailed('exec-1', new Error('boom'));
+
+			await expect(wait).rejects.toThrow('boom');
+		});
+
+		it('should reject at once when the worker reported the failure before the wait started', async () => {
+			tracker.recordFailed('exec-1', new Error('boom'));
+
+			await expect(tracker.waitFor(job)).rejects.toThrow('boom');
+		});
+
+		it('should not keep an early failure for an execution this process did not enqueue', async () => {
+			activeExecutions.has.mockReturnValue(false);
+			tracker.recordFailed('exec-1', new Error('boom'));
+			activeExecutions.has.mockReturnValue(true);
+
+			const wait = tracker.waitFor(job);
+			tracker.recordFinished('exec-1', result);
+
+			await expect(wait).resolves.toBeUndefined();
+		});
+
+		it('should settle from a Bull event by queue name and job ID', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.settleByJobKey('jobs', 'job-1', new Error('job stalled more than maxStalledCount'));
+
+			await expect(wait).rejects.toThrow('job stalled more than maxStalledCount');
+		});
+
+		it('should settle only the wait for the queue that emitted the Bull event', async () => {
+			const poolJob = mock<Job>({
+				id: 'job-1',
+				data: { executionId: 'exec-pool' },
+				queue: { name: 'jobs-gpu' },
+			});
+			let poolSettled = false;
+			void tracker.waitFor(poolJob).finally(() => (poolSettled = true));
+			const wait = tracker.waitFor(job);
+
+			// Bull job IDs are unique per queue only
+			tracker.settleByJobKey('jobs', 'job-1');
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(poolSettled).toBe(false);
+		});
+	});
+
+	describe('response promise', () => {
+		it('should answer the request with success when the job ended well', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.settleByJobKey('jobs', 'job-1');
+			await wait;
+
+			expect(activeExecutions.resolveResponsePromise).toHaveBeenCalledWith('exec-1', {});
+		});
+
+		it('should answer the request with an error when the job failed', async () => {
+			const wait = tracker.waitFor(job);
+
+			tracker.recordFailed('exec-1', new Error('boom'));
+			await expect(wait).rejects.toThrow();
+
+			expect(activeExecutions.resolveResponsePromise).toHaveBeenCalledWith(
+				'exec-1',
+				expect.objectContaining({ statusCode: 500 }),
+			);
+		});
+	});
+
+	describe('recheck', () => {
+		it('should resolve from the DB when every completion event was missed', async () => {
+			vi.useFakeTimers();
+			executionRepository.findStatusesByIds.mockResolvedValue(statusRows({ 'exec-1': 'success' }));
+
+			const wait = tracker.waitFor(job);
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS);
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(activeExecutions.resolveResponsePromise).toHaveBeenCalledWith('exec-1', {});
+			expect(eventService.emit).toHaveBeenCalledWith('job-completion-missed', {
+				status: 'success',
+			});
+		});
+
+		it('should answer the request with an error when the DB shows the execution failed', async () => {
+			vi.useFakeTimers();
+			executionRepository.findStatusesByIds.mockResolvedValue(statusRows({ 'exec-1': 'error' }));
+
+			const wait = tracker.waitFor(job);
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS);
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(activeExecutions.resolveResponsePromise).toHaveBeenCalledWith(
+				'exec-1',
+				expect.objectContaining({ statusCode: 500 }),
+			);
+		});
+
+		it('should resolve from the DB when the execution row is gone', async () => {
+			vi.useFakeTimers();
+			executionRepository.findStatusesByIds.mockResolvedValue([]);
+
+			const wait = tracker.waitFor(job);
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS);
+
+			await expect(wait).resolves.toBeUndefined();
+		});
+
+		it('should keep waiting while the DB still shows the execution as running', async () => {
+			vi.useFakeTimers();
+			executionRepository.findStatusesByIds.mockResolvedValue(statusRows({ 'exec-1': 'running' }));
+			let settled = false;
+
+			void tracker.waitFor(job).finally(() => (settled = true));
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS * 2);
+
+			expect(executionRepository.findStatusesByIds).toHaveBeenCalledTimes(2);
+			expect(settled).toBe(false);
+		});
+
+		it('should keep rechecking after a failed DB read', async () => {
+			vi.useFakeTimers();
+			executionRepository.findStatusesByIds
+				.mockRejectedValueOnce(new Error('db unavailable'))
+				.mockResolvedValue(statusRows({ 'exec-1': 'success' }));
+
+			const wait = tracker.waitFor(job);
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS * 2);
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(executionRepository.findStatusesByIds).toHaveBeenCalledTimes(2);
+		});
+
+		it('should read the status of all pending waits in one query', async () => {
+			vi.useFakeTimers();
+			const otherJob = mock<Job>({
+				id: 'job-2',
+				data: { executionId: 'exec-2' },
+				queue: { name: 'jobs' },
+			});
+			executionRepository.findStatusesByIds.mockResolvedValue(
+				statusRows({ 'exec-1': 'success', 'exec-2': 'running' }),
+			);
+			let otherSettled = false;
+
+			const wait = tracker.waitFor(job);
+			void tracker.waitFor(otherJob).finally(() => (otherSettled = true));
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS);
+
+			expect(executionRepository.findStatusesByIds).toHaveBeenCalledTimes(1);
+			expect(executionRepository.findStatusesByIds).toHaveBeenCalledWith(['exec-1', 'exec-2']);
+			await expect(wait).resolves.toBeUndefined();
+			expect(otherSettled).toBe(false);
+		});
+
+		it('should not count a wait that an event settled while the DB read was in flight', async () => {
+			let releaseRead: (rows: Array<{ id: string; status: ExecutionStatus }>) => void = () => {};
+			executionRepository.findStatusesByIds.mockReturnValue(
+				new Promise((resolve) => (releaseRead = resolve)),
+			);
+
+			const wait = tracker.waitFor(job);
+			const recheck = tracker.recheckAll();
+			tracker.recordFinished('exec-1', result);
+			releaseRead(statusRows({ 'exec-1': 'success' }));
+			await recheck;
+
+			await expect(wait).resolves.toBeUndefined();
+			expect(eventService.emit).not.toHaveBeenCalled();
+		});
+
+		it('should join a recheck that is already reading the DB instead of starting another', async () => {
+			let releaseRead: (rows: Array<{ id: string; status: ExecutionStatus }>) => void = () => {};
+			executionRepository.findStatusesByIds.mockReturnValue(
+				new Promise((resolve) => (releaseRead = resolve)),
+			);
+
+			const wait = tracker.waitFor(job);
+			// A reconnect during a slow timer tick, or the other way round
+			const first = tracker.recheckAll();
+			const second = tracker.recheckAll();
+			releaseRead(statusRows({ 'exec-1': 'success' }));
+			await Promise.all([first, second]);
+
+			expect(executionRepository.findStatusesByIds).toHaveBeenCalledTimes(1);
+			await expect(wait).resolves.toBeUndefined();
+		});
+
+		it('should recheck at once on demand, e.g. when Redis reconnects', async () => {
+			executionRepository.findStatusesByIds.mockResolvedValue(statusRows({ 'exec-1': 'success' }));
+
+			const wait = tracker.waitFor(job);
+			await tracker.recheckAll();
+
+			await expect(wait).resolves.toBeUndefined();
+		});
+
+		it('should not query the DB when nothing is pending', async () => {
+			await tracker.recheckAll();
+
+			expect(executionRepository.findStatusesByIds).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('drop', () => {
+		it('should stop the recheck once no wait is pending and leave the wait unsettled', async () => {
+			vi.useFakeTimers();
+			let settled = false;
+
+			void tracker.waitFor(job).finally(() => (settled = true));
+			tracker.drop('exec-1');
+			await vi.advanceTimersByTimeAsync(JOB_WAIT_RECHECK_INTERVAL_MS * 2);
+
+			expect(executionRepository.findStatusesByIds).not.toHaveBeenCalled();
+			expect(settled).toBe(false);
+		});
+	});
+});

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

@@ -6,7 +6,7 @@ import { Container } from '@n8n/di';
 import * as BullModule from 'bull';
 import { ENCODED_BUFFER_KEY, InstanceSettings } from 'n8n-core';
 import type { ErrorReporter } from 'n8n-core';
-import { UnexpectedError } from 'n8n-workflow';
+import { OperationalError, UnexpectedError } from 'n8n-workflow';
 import type { MockInstance } from 'vitest';
 import { mock } from 'vitest-mock-extended';
 
@@ -15,12 +15,14 @@ import { ExecutionCrashService } from '@/executions/execution-crash.service';
 import type { ExecutionPersistence } from '@/executions/execution-persistence';
 
 import { JOB_TYPE_NAME } from '../constants';
+import { JobOutcomeTracker } from '../job-outcome-tracker';
 import type { JobProcessor } from '../job-processor';
 import { ScalingService } from '../scaling.service';
 import type { Job, JobData, JobId, JobQueue } from '../scaling.types';
 import type { WebhookResponseRelay } from '../webhook-response-relay';
 
 const queue = mock<JobQueue>({
+	name: 'jobs',
 	client: { ping: vi.fn() },
 });
 
@@ -102,6 +104,7 @@ describe('ScalingService', () => {
 	const executionRepository = mock<ExecutionRepository>();
 	const executionPersistence = mock<ExecutionPersistence>();
 	const executionCrashService = mockInstance(ExecutionCrashService);
+	const jobOutcomeTracker = mock<JobOutcomeTracker>();
 	const webhookResponseRelay = mock<WebhookResponseRelay>();
 
 	let scalingService: ScalingService;
@@ -146,6 +149,7 @@ describe('ScalingService', () => {
 			mock(),
 			webhookResponseRelay,
 			executionCrashService,
+			jobOutcomeTracker,
 		);
 
 		getRunningJobsCountSpy = vi.spyOn(scalingService, 'getRunningJobsCount');
@@ -174,6 +178,19 @@ describe('ScalingService', () => {
 				expect(registerWorkerListenersSpy).not.toHaveBeenCalled();
 				expect(scheduleQueueRecoverySpy).toHaveBeenCalledWith(0);
 			});
+
+			it('should recheck pending job waits when the Redis connection recovers', async () => {
+				await scalingService.setupQueue();
+				const { RedisClientService } = await import('@n8n/backend-services');
+
+				// Completion events sent while the connection was down are lost
+				// The service debounces its emits, so the event lands on the next second
+				vi.useFakeTimers();
+				Container.get(RedisClientService).emit('connection-recovered');
+				await vi.advanceTimersByTimeAsync(1000);
+
+				expect(jobOutcomeTracker.recheckAll).toHaveBeenCalled();
+			});
 		});
 
 		describe('if follower main', () => {
@@ -338,6 +355,17 @@ describe('ScalingService', () => {
 				expect(stopQueueRecoverySpy).toHaveBeenCalled();
 				expect(stopQueueMetricsSpy).toHaveBeenCalled();
 			});
+
+			it('should keep pending job waits so the active executions drain can settle them', async () => {
+				// @ts-expect-error readonly property
+				instanceSettings.instanceType = 'main';
+				await scalingService.setupQueue();
+
+				await scalingService.stop();
+
+				expect(jobOutcomeTracker.clear).not.toHaveBeenCalled();
+				expect(jobOutcomeTracker.drop).not.toHaveBeenCalled();
+			});
 		});
 
 		describe('if worker', () => {
@@ -739,6 +767,7 @@ describe('ScalingService', () => {
 				mock(),
 				webhookResponseRelay,
 				executionCrashService,
+				jobOutcomeTracker,
 			);
 
 			await scalingService.setupQueue();
@@ -778,6 +807,7 @@ describe('ScalingService', () => {
 				mock(),
 				webhookResponseRelay,
 				executionCrashService,
+				jobOutcomeTracker,
 			);
 
 			await scalingService.setupQueue();
@@ -812,6 +842,7 @@ describe('ScalingService', () => {
 				mock(),
 				webhookResponseRelay,
 				executionCrashService,
+				jobOutcomeTracker,
 			);
 
 			await scalingService.setupQueue();
@@ -834,91 +865,126 @@ describe('ScalingService', () => {
 				statusCode: 500,
 			});
 		});
+		describe('job outcome wiring', () => {
+			const getHandler = (event: string) =>
+				queue.on.mock.calls.find(([name]) => (name as string) === event)?.[1] as (
+					...args: unknown[]
+				) => void;
 
-		it('should keep waitTill when storing a v2 job-finished result', async () => {
-			const activeExecutions = mock<ActiveExecutions>();
-			activeExecutions.has.mockReturnValue(true);
-			scalingService = new ScalingService(
-				mockLogger(),
-				mock(),
-				activeExecutions,
-				jobProcessor,
-				globalConfig,
-				mock(),
-				mock(),
-				instanceSettings,
-				mock(),
-				webhookResponseRelay,
-				executionCrashService,
-			);
-
-			await scalingService.setupQueue();
-
-			const messageHandler = queue.on.mock.calls.find(
-				([event]) => (event as string) === 'global:progress',
-			)?.[1] as (jobId: JobId, msg: unknown) => void;
-
-			const waitTill = new Date('2026-07-25T12:00:00.000Z');
-			// Bull delivers progress messages JSON-serialized, so dates arrive as ISO strings
-			const jobFinishedMessage = {
-				kind: 'job-finished',
-				version: 2,
-				executionId: 'exec-123',
-				workerId: 'worker-456',
-				success: true,
-				status: 'waiting',
-				startedAt: '2026-07-25T11:59:00.000Z',
-				stoppedAt: '2026-07-25T11:59:30.000Z',
-				waitTill: waitTill.toISOString(),
-			};
-
-			messageHandler('job-789', jobFinishedMessage);
-
-			const result = scalingService.popJobResult('exec-123');
-
-			expect(result?.status).toBe('waiting');
-			// A missing waitTill makes main treat a waiting execution as finished and
-			// delete it when the workflow does not save successful executions
-			expect(result?.waitTill).toEqual(waitTill);
-		});
-
-		it('should not store a job-finished result for an execution this process did not enqueue', async () => {
-			const activeExecutions = mock<ActiveExecutions>();
-			activeExecutions.has.mockReturnValue(false);
-			scalingService = new ScalingService(
-				mockLogger(),
-				mock(),
-				activeExecutions,
-				jobProcessor,
-				globalConfig,
-				mock(),
-				mock(),
-				instanceSettings,
-				mock(),
-				webhookResponseRelay,
-				executionCrashService,
-			);
-
-			await scalingService.setupQueue();
-
-			const messageHandler = queue.on.mock.calls.find(
-				([event]) => (event as string) === 'global:progress',
-			)?.[1] as (jobId: JobId, msg: unknown) => void;
-
-			// Bull broadcasts progress messages to every main and webhook process
-			messageHandler('job-789', {
-				kind: 'job-finished',
-				version: 2,
-				executionId: 'exec-other-main',
-				workerId: 'worker-456',
-				success: true,
-				status: 'success',
-				startedAt: '2026-07-25T11:59:00.000Z',
-				stoppedAt: '2026-07-25T11:59:30.000Z',
+			beforeEach(async () => {
+				await scalingService.setupQueue();
 			});
 
-			expect(activeExecutions.has).toHaveBeenCalledWith('exec-other-main');
-			expect(scalingService.popJobResult('exec-other-main')).toBeUndefined();
+			it('should record a v2 job-finished result with its dates and waitTill', () => {
+				const waitTill = new Date('2026-07-25T12:00:00.000Z');
+				// Bull delivers progress messages JSON-serialized, so dates arrive as ISO strings
+				getHandler('global:progress')('job-789', {
+					kind: 'job-finished',
+					version: 2,
+					executionId: 'exec-123',
+					workerId: 'worker-456',
+					success: true,
+					status: 'waiting',
+					startedAt: '2026-07-25T11:59:00.000Z',
+					stoppedAt: '2026-07-25T11:59:30.000Z',
+					waitTill: waitTill.toISOString(),
+				});
+
+				expect(jobOutcomeTracker.recordFinished).toHaveBeenCalledWith(
+					'exec-123',
+					expect.objectContaining({
+						status: 'waiting',
+						startedAt: new Date('2026-07-25T11:59:00.000Z'),
+						// A missing waitTill makes main treat a waiting execution as finished and
+						// delete it when the workflow does not save successful executions
+						waitTill,
+					}),
+				);
+			});
+
+			it('should record a v1 job-finished message without a result', () => {
+				getHandler('global:progress')('job-789', {
+					kind: 'job-finished',
+					executionId: 'exec-123',
+					workerId: 'worker-456',
+					success: true,
+				});
+
+				expect(jobOutcomeTracker.recordFinished).toHaveBeenCalledWith('exec-123', undefined);
+			});
+
+			it('should record a job-failed report as a handled error', () => {
+				getHandler('global:progress')('job-789', {
+					kind: 'job-failed',
+					executionId: 'exec-123',
+					workerId: 'worker-456',
+					errorMsg: 'boom',
+					errorStack: '',
+				});
+
+				expect(jobOutcomeTracker.recordFailed).toHaveBeenCalledWith(
+					'exec-123',
+					expect.any(OperationalError),
+				);
+				expect(jobOutcomeTracker.recordFailed.mock.calls[0][1].message).toBe('boom');
+			});
+
+			it('should settle the wait for a job Bull reports as failed, by queue and job ID', () => {
+				getHandler('global:failed')('job-1', 'job stalled more than maxStalledCount');
+
+				expect(jobOutcomeTracker.settleByJobKey).toHaveBeenCalledWith(
+					'jobs',
+					'job-1',
+					expect.any(OperationalError),
+				);
+			});
+
+			it('should settle the wait for a job Bull reports as completed, by queue and job ID', () => {
+				getHandler('global:completed')('job-1');
+
+				expect(jobOutcomeTracker.settleByJobKey).toHaveBeenCalledWith('jobs', 'job-1');
+			});
+
+			it('should end the wait when an older worker reports the job as finished', async () => {
+				// A real tracker, so the handler and the tracker are checked together
+				const realTracker = new JobOutcomeTracker(mockLogger(), activeExecutions, mock(), mock());
+				const service = new ScalingService(
+					logger,
+					errorReporter,
+					activeExecutions,
+					jobProcessor,
+					globalConfig,
+					executionRepository,
+					executionPersistence,
+					instanceSettings,
+					mock(),
+					webhookResponseRelay,
+					executionCrashService,
+					realTracker,
+				);
+				await service.setupQueue();
+				const onProgress = queue.on.mock.calls
+					.filter(([event]) => (event as string) === 'global:progress')
+					.at(-1)?.[1] as (jobId: JobId, msg: unknown) => void;
+
+				const job = mock<Job>({
+					id: 'job-1',
+					data: { executionId: 'exec-1' },
+					queue: { name: 'jobs' },
+				});
+				const wait = service.waitForJob(job);
+
+				// A v1 message carries no result, only the fact that the job ended
+				onProgress('job-1', {
+					kind: 'job-finished',
+					executionId: 'exec-1',
+					workerId: 'worker-1',
+					success: true,
+				});
+
+				await expect(wait).resolves.toBeUndefined();
+				expect(service.popJobResult('exec-1')).toBeUndefined();
+			});
 		});
 	});
 

packages/cli/src/scaling/constants.ts

@@ -69,3 +69,6 @@ export const IMMEDIATE_COMMANDS = new Set<PubSub.Command['command']>([
 	'display-workflow-publication-status',
 	'workflow-publish-wake-up',
 ]);
+
+/** How often a main rechecks the DB for a queued job whose completion event it may have missed. */
+export const JOB_WAIT_RECHECK_INTERVAL_MS = 60 * Time.seconds.toMilliseconds;

packages/cli/src/scaling/job-outcome-tracker.ts

@@ -0,0 +1,253 @@
+import { Logger } from '@n8n/backend-common';
+import { EventService } from '@n8n/backend-services';
+import { ExecutionRepository } from '@n8n/db';
+import { Service } from '@n8n/di';
+import type { ExecutionStatus, IExecuteResponsePromiseData } from 'n8n-workflow';
+
+import { ActiveExecutions } from '@/active-executions';
+
+import { JOB_WAIT_RECHECK_INTERVAL_MS } from './constants';
+import type { Job, JobFinishedProps, JobId } from './scaling.types';
+
+type PendingJobWait = {
+	jobKey: string;
+	resolve: () => void;
+	reject: (error: Error) => void;
+};
+
+type JobOutcome = {
+	error?: Error;
+	/** Whether the request that started the job should get a success response. */
+	succeeded: boolean;
+};
+
+/** Bull job IDs are unique per queue only, so pool queues can reuse them. */
+const toJobKey = (queueName: string, jobId: JobId) => `${queueName}:${jobId.toString()}`;
+
+type RecheckedStatus = ExecutionStatus | 'deleted';
+
+/** `unknown` is stoppable and in flight elsewhere in the codebase, so it is in flight here too. */
+const IN_FLIGHT_STATUSES = new Set<RecheckedStatus>(['new', 'running', 'unknown']);
+
+/** A missing row means the worker finished and the execution was not saved, so nothing failed. */
+const SUCCEEDED_STATUSES = new Set<RecheckedStatus>(['success', 'waiting', 'deleted']);
+
+const FAILED_RESPONSE: IExecuteResponsePromiseData = {
+	body: { message: 'Workflow execution failed' },
+	statusCode: 500,
+};
+
+/**
+ * Tracks the outcome of queued jobs on a `main` or `webhook` process, and lets
+ * the process that enqueued a job wait for it.
+ *
+ * Bull's `job.finished()` is not used because it never settles when the
+ * completion event is missed: `removeOnComplete` deletes the job before
+ * Bull's poll can see it, so the poll, its listeners and the caller's
+ * closure stay alive until restart. Here a missed event is covered by a
+ * slow recheck of the execution status in the DB, and by a recheck as soon
+ * as the Redis connection recovers.
+ */
+@Service()
+export class JobOutcomeTracker {
+	/** Results the worker reported for jobs this process enqueued, keyed by execution ID. */
+	private readonly results = new Map<string, JobFinishedProps>();
+
+	/** Failures the worker reported before this process started waiting, keyed by execution ID. */
+	private readonly failures = new Map<string, Error>();
+
+	/** Waits for jobs to end, keyed by execution ID. */
+	private readonly pendingWaits = new Map<string, PendingJobWait>();
+
+	/** Execution ID for each job being waited for, keyed by queue name and job ID. */
+	private readonly executionIdByJobKey = new Map<string, string>();
+
+	/** One timer for all pending waits, running only while there are any. */
+	private recheckTimer?: NodeJS.Timeout;
+
+	/** The recheck currently reading the DB, so a timer tick and a reconnect do not overlap. */
+	private recheckInFlight?: Promise<void>;
+
+	constructor(
+		private readonly logger: Logger,
+		private readonly activeExecutions: ActiveExecutions,
+		private readonly executionRepository: ExecutionRepository,
+		private readonly eventService: EventService,
+	) {
+		this.logger = this.logger.scoped('scaling');
+	}
+
+	/**
+	 * Record that the worker reported the job as finished, with the result when
+	 * the worker sent one. Bull broadcasts the message to every main and webhook
+	 * process, but only the process that enqueued the job ever pops the result.
+	 */
+	recordFinished(executionId: string, result?: JobFinishedProps) {
+		if (result && this.activeExecutions.has(executionId)) {
+			this.results.set(executionId, result);
+		}
+
+		this.settle(executionId, { succeeded: result?.success ?? true });
+	}
+
+	/** Record a failure the worker reported, or reject the wait for it at once. */
+	recordFailed(executionId: string, error: Error) {
+		const settled = this.settle(executionId, { error, succeeded: false });
+		if (settled) return;
+
+		// A fast failure can arrive before the enqueuing process starts to wait
+		if (this.activeExecutions.has(executionId)) this.failures.set(executionId, error);
+	}
+
+	/** Settle the wait for a job from a Bull event, which carries only the job ID. */
+	settleByJobKey(queueName: string, jobId: JobId, error?: Error) {
+		const executionId = this.executionIdByJobKey.get(toJobKey(queueName, jobId));
+		if (!executionId) return;
+
+		this.settle(executionId, { error, succeeded: !error });
+	}
+
+	/**
+	 * Wait until the worker reports the job as finished, or Bull reports it as
+	 * failed. Rejects with the failure reason, like Bull's `job.finished()`.
+	 */
+	async waitFor(job: Job): Promise<void> {
+		const { executionId } = job.data;
+
+		// The worker may have reported the outcome before this wait was registered.
+		if (this.results.has(executionId)) return;
+
+		const earlyFailure = this.failures.get(executionId);
+		if (earlyFailure) {
+			this.failures.delete(executionId);
+			throw earlyFailure;
+		}
+
+		await new Promise<void>((resolve, reject) => {
+			const jobKey = toJobKey(job.queue.name, job.id);
+			this.pendingWaits.set(executionId, { jobKey, resolve, reject });
+			this.executionIdByJobKey.set(jobKey, executionId);
+			this.startRecheckTimer();
+		});
+	}
+
+	/** Get and remove the result for a finished job. */
+	popResult(executionId: string): JobFinishedProps | undefined {
+		const result = this.results.get(executionId);
+		this.results.delete(executionId);
+		this.failures.delete(executionId);
+		return result;
+	}
+
+	/** Drop the wait without settling it, e.g. when the caller cancelled the execution. */
+	drop(executionId: string) {
+		const wait = this.pendingWaits.get(executionId);
+		if (!wait) return undefined;
+
+		this.pendingWaits.delete(executionId);
+		this.executionIdByJobKey.delete(wait.jobKey);
+		if (this.pendingWaits.size === 0) this.stopRecheckTimer();
+
+		return wait;
+	}
+
+	/** Drop every wait. */
+	clear() {
+		for (const executionId of this.pendingWaits.keys()) this.drop(executionId);
+	}
+
+	/**
+	 * Settle every wait whose completion event was missed, once the DB shows the
+	 * execution ended. Runs on the timer, and at once when Redis reconnects.
+	 * Only one recheck runs at a time; a call during a recheck joins it.
+	 */
+	async recheckAll() {
+		if (this.pendingWaits.size === 0) return;
+
+		this.recheckInFlight ??= this.recheckPendingWaits().finally(() => {
+			this.recheckInFlight = undefined;
+		});
+
+		await this.recheckInFlight;
+	}
+
+	private async recheckPendingWaits() {
+		const statusById = await this.readStatuses([...this.pendingWaits.keys()]);
+		if (!statusById) return;
+
+		for (const [executionId, status] of statusById) {
+			if (IN_FLIGHT_STATUSES.has(status)) continue;
+			// An event may have settled the wait while the DB read was in flight
+			if (!this.pendingWaits.has(executionId)) continue;
+
+			this.logger.warn(
+				`Execution ${executionId} ended without a completion event, resolving the wait from the DB`,
+				{ executionId, status },
+			);
+			this.eventService.emit('job-completion-missed', { status });
+			this.settle(executionId, { succeeded: SUCCEEDED_STATUSES.has(status) });
+		}
+	}
+
+	/**
+	 * @returns the status of each execution, `deleted` when its row is gone, or
+	 * `undefined` when the read failed. The recheck interval is the retry.
+	 */
+	private async readStatuses(
+		executionIds: string[],
+	): Promise<Map<string, RecheckedStatus> | undefined> {
+		try {
+			const rows = await this.executionRepository.findStatusesByIds(executionIds);
+			const statusById = new Map<string, RecheckedStatus>(rows.map((row) => [row.id, row.status]));
+
+			for (const executionId of executionIds) {
+				if (!statusById.has(executionId)) statusById.set(executionId, 'deleted');
+			}
+
+			return statusById;
+		} catch (error) {
+			this.logger.warn(
+				`Failed to read the status of ${executionIds.length} executions, next recheck in ${JOB_WAIT_RECHECK_INTERVAL_MS / 1000}s`,
+				{ executionIds, error },
+			);
+			return undefined;
+		}
+	}
+
+	/** @returns whether a wait was pending for this execution */
+	private settle(executionId: string, outcome: JobOutcome) {
+		const wait = this.drop(executionId);
+		if (!wait) return false;
+
+		// The request may still wait for a response the worker sent while this process
+		// was disconnected. Resolving twice is a no-op.
+		this.activeExecutions.resolveResponsePromise(
+			executionId,
+			outcome.succeeded ? {} : FAILED_RESPONSE,
+		);
+
+		if (outcome.error) {
+			wait.reject(outcome.error);
+		} else {
+			wait.resolve();
+		}
+
+		return true;
+	}
+
+	private startRecheckTimer() {
+		if (this.recheckTimer) return;
+
+		// Unref'd so pending waits do not hold the process open during shutdown
+		this.recheckTimer = setInterval(() => {
+			void this.recheckAll();
+		}, JOB_WAIT_RECHECK_INTERVAL_MS).unref();
+	}
+
+	private stopRecheckTimer() {
+		if (!this.recheckTimer) return;
+
+		clearInterval(this.recheckTimer);
+		this.recheckTimer = undefined;
+	}
+}

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

@@ -8,7 +8,7 @@ import { Container, Service } from '@n8n/di';
 import { decodeBufferBody, ErrorReporter, InstanceSettings } from 'n8n-core';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import { sleep } from '@n8n/utils/sleep';
-import { jsonStringify, UnexpectedError } from 'n8n-workflow';
+import { jsonStringify, OperationalError, UnexpectedError } from 'n8n-workflow';
 import type { IRun } from 'n8n-workflow';
 import assert, { strict } from 'node:assert';
 
@@ -19,6 +19,7 @@ import { ExecutionPersistence } from '@/executions/execution-persistence';
 import { assertNever } from '@/utils';
 
 import { JOB_TYPE_NAME } from './constants';
+import { JobOutcomeTracker } from './job-outcome-tracker';
 import { JobProcessor } from './job-processor';
 import { DEFAULT_QUEUE_NAME, resolveQueueName, resolveWorkerPoolName } from './queue-name';
 import type {
@@ -52,8 +53,6 @@ export class ScalingService {
 
 	private createBullQueue?: (name: string) => JobQueue;
 
-	private jobResults = new Map<string, JobFinishedProps>();
-
 	constructor(
 		private readonly logger: Logger,
 		private readonly errorReporter: ErrorReporter,
@@ -66,6 +65,7 @@ export class ScalingService {
 		private readonly eventService: EventService,
 		private readonly webhookResponseRelay: WebhookResponseRelay,
 		private readonly executionCrashService: ExecutionCrashService,
+		private readonly jobOutcomeTracker: JobOutcomeTracker,
 	) {
 		this.logger = this.logger.scoped('scaling');
 	}
@@ -102,6 +102,11 @@ export class ScalingService {
 
 		const service = Container.get(RedisClientService);
 
+		// Completion events sent while the connection was down are lost, so check the DB at once
+		service.on('connection-recovered', () => {
+			void this.jobOutcomeTracker.recheckAll();
+		});
+
 		const bullPrefix = this.globalConfig.queue.bull.prefix;
 		const prefix = service.toValidPrefix(bullPrefix);
 		const settings = { ...this.globalConfig.queue.bull.settings, maxStalledCount: 0 };
@@ -336,9 +341,12 @@ export class ScalingService {
 
 	/** Get and remove the result for a completed job. */
 	popJobResult(executionId: string): JobFinishedProps | undefined {
-		const result = this.jobResults.get(executionId);
-		this.jobResults.delete(executionId);
-		return result;
+		return this.jobOutcomeTracker.popResult(executionId);
+	}
+
+	/** Wait until the job ends. Rejects with the failure reason, like Bull's `job.finished()`. */
+	async waitForJob(job: Job): Promise<void> {
+		await this.jobOutcomeTracker.waitFor(job);
 	}
 
 	async getPendingJobCounts() {
@@ -409,6 +417,9 @@ export class ScalingService {
 	async stopJob(job: Job) {
 		const props = { jobId: job.id, executionId: job.data.executionId };
 
+		// A removed job emits no completion event, and the caller handles the cancellation
+		this.jobOutcomeTracker.drop(job.data.executionId);
+
 		try {
 			if (await job.isActive()) {
 				await job.progress({ kind: 'abort-job' }); // being processed by worker
@@ -523,25 +534,28 @@ export class ScalingService {
 					 * because `removeOnComplete: true` prevents `job.finished()`
 					 * from returning a value that is no longer in Redis.
 					 *
-					 * Bull broadcasts this message to every main and webhook process,
-					 * but only the process that enqueued the job ever pops the result.
+					 * A v1 message from an older worker carries no result, but it
+					 * still ends the wait for the job.
 					 */
-					if (msg.version === 2 && this.activeExecutions.has(msg.executionId)) {
-						this.jobResults.set(msg.executionId, {
-							success: msg.success,
-							error: msg.error,
-							status: msg.status,
-							lastNodeExecuted: msg.lastNodeExecuted,
-							usedDynamicCredentials: msg.usedDynamicCredentials,
-							metadata: msg.metadata,
-							startedAt: new Date(msg.startedAt),
-							stoppedAt: new Date(msg.stoppedAt),
-							// Dropping `waitTill` here makes main mistake a waiting execution
-							// for a finished one and delete it when the workflow does not
-							// save successful executions
-							waitTill: msg.waitTill ? new Date(msg.waitTill) : null,
-						});
-					}
+					this.jobOutcomeTracker.recordFinished(
+						msg.executionId,
+						msg.version === 2
+							? {
+									success: msg.success,
+									error: msg.error,
+									status: msg.status,
+									lastNodeExecuted: msg.lastNodeExecuted,
+									usedDynamicCredentials: msg.usedDynamicCredentials,
+									metadata: msg.metadata,
+									startedAt: new Date(msg.startedAt),
+									stoppedAt: new Date(msg.stoppedAt),
+									// Dropping `waitTill` here makes main mistake a waiting execution
+									// for a finished one and delete it when the workflow does not
+									// save successful executions
+									waitTill: msg.waitTill ? new Date(msg.waitTill) : null,
+								}
+							: undefined,
+					);
 
 					this.logger.info(`Execution ${msg.executionId} (job ${jobId}) finished`, {
 						workerId: msg.workerId,
@@ -563,6 +577,9 @@ export class ScalingService {
 							jobId,
 						},
 					);
+
+					// The worker already reported the underlying error, so this copy is only a handled signal
+					this.jobOutcomeTracker.recordFailed(msg.executionId, new OperationalError(msg.errorMsg));
 					break;
 				case 'abort-job':
 					break; // only for worker
@@ -582,6 +599,14 @@ export class ScalingService {
 			}
 		});
 
+		// Failures such as a stall are reported only by Bull, not by the worker
+		queue.on('global:failed', (jobId: JobId, failedReason: string) => {
+			this.jobOutcomeTracker.settleByJobKey(queue.name, jobId, new OperationalError(failedReason));
+		});
+		queue.on('global:completed', (jobId: JobId) => {
+			this.jobOutcomeTracker.settleByJobKey(queue.name, jobId);
+		});
+
 		if (this.isQueueMetricsEnabled) {
 			queue.on('global:completed', () => this.jobCounters.completed++);
 			queue.on('global:failed', () => this.jobCounters.failed++);

packages/cli/src/workflow-runner.ts

@@ -5,6 +5,7 @@ import { EventService } from '@n8n/backend-services';
 import { ExecutionsConfig } from '@n8n/config';
 import { Time } from '@n8n/constants';
 import { ExecutionRepository } from '@n8n/db';
+import type { IExecutionResponse } from '@n8n/db';
 import { Container, Service } from '@n8n/di';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
@@ -796,6 +797,8 @@ export class WorkflowRunner {
 			// "workflowExecuteAfter" which we require.
 			const lifecycleHooks = getLifecycleHooksForScalingWorker(data, executionId);
 			await this.processError(error, new Date(), data.executionMode, executionId, lifecycleHooks);
+			// Nobody will wait for this job, so drop any outcome the worker already reported
+			this.scalingService.popJobResult(executionId);
 			throw error;
 		}
 
@@ -821,7 +824,7 @@ export class WorkflowRunner {
 				});
 
 				try {
-					await job.finished();
+					await this.scalingService.waitForJob(job);
 				} catch (error) {
 					if (
 						error instanceof Error &&
@@ -862,15 +865,43 @@ export class WorkflowRunner {
 					!jobResult ||
 					this.needsFullExecutionData(data.executionMode, executionId, data.forceFullExecutionData)
 				) {
-					const fullExecutionData = await this.executionPersistence.findSingleExecution(
-						executionId,
-						{
+					let fullExecutionData: IExecutionResponse | undefined;
+					try {
+						fullExecutionData = await this.executionPersistence.findSingleExecution(executionId, {
 							includeData: true,
 							unflattenData: true,
-						},
-					);
+						});
+					} catch (error) {
+						// An async executor's throw would never settle this promise, and the
+						// active execution would keep the request alive until restart
+						await this.processError(
+							error,
+							new Date(),
+							data.executionMode,
+							executionId,
+							getLifecycleHooksForScalingWorker(data, executionId),
+						);
+						return reject(error);
+					}
+
 					if (!fullExecutionData) {
-						return reject(new Error(`Could not find execution with id "${executionId}"`));
+						// Not a bug by itself: the worker deletes an execution that is not saved.
+						// Finalizing with a failed run makes the webhook respond with an error
+						// instead of a success without data.
+						this.logger.warn(`Execution ${executionId} ended but its record is gone`, {
+							executionId,
+						});
+						const error = new WorkflowOperationError(
+							`Could not find execution with id "${executionId}"`,
+						);
+						await this.processError(
+							error,
+							new Date(),
+							data.executionMode,
+							executionId,
+							getLifecycleHooksForScalingWorker(data, executionId),
+						);
+						return reject(error);
 					}
 
 					runData = {

packages/cli/test/integration/scaling/queue-recovery.test.ts

@@ -42,6 +42,7 @@ describe('ScalingService queue recovery', () => {
 			mock(),
 			mock(),
 			Container.get(ExecutionCrashService),
+			mock(),
 		);
 	});