fix(core): Ensure job processor does not reprocess amended executions (#11438)

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

@@ -0,0 +1,21 @@
+import { mock } from 'jest-mock-extended';
+
+import type { ExecutionRepository } from '@/databases/repositories/execution.repository';
+import type { IExecutionResponse } from '@/interfaces';
+
+import { JobProcessor } from '../job-processor';
+import type { Job } from '../scaling.types';
+
+describe('JobProcessor', () => {
+	it('should refrain from processing a crashed execution', async () => {
+		const executionRepository = mock<ExecutionRepository>();
+		executionRepository.findSingleExecution.mockResolvedValue(
+			mock<IExecutionResponse>({ status: 'crashed' }),
+		);
+		const jobProcessor = new JobProcessor(mock(), executionRepository, mock(), mock(), mock());
+
+		const result = await jobProcessor.processJob(mock<Job>());
+
+		expect(result).toEqual({ success: false });
+	});
+});

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

@@ -58,6 +58,13 @@ export class JobProcessor {
 			);
 		}
 
+		/**
+		 * Bull's implicit retry mechanism and n8n's execution recovery mechanism may
+		 * cause a crashed execution to be enqueued. We refrain from processing it,
+		 * until we have reworked both mechanisms to prevent this scenario.
+		 */
+		if (execution.status === 'crashed') return { success: false };
+
 		const workflowId = execution.workflowData.id;
 
 		this.logger.info(`Worker started execution ${executionId} (job ${job.id})`, {