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})`, {

packages/cli/src/workflow-runner.ts

@@ -14,7 +14,6 @@ import type {
 	IWorkflowExecutionDataProcess,
 } from 'n8n-workflow';
 import {
-	ApplicationError,
 	ErrorReporterProxy as ErrorReporter,
 	ExecutionCancelledError,
 	Workflow,
@@ -381,17 +380,6 @@ export class WorkflowRunner {
 		let job: Job;
 		let hooks: WorkflowHooks;
 		try {
-			// check to help diagnose PAY-2100
-			if (
-				data.executionData?.executionData?.nodeExecutionStack?.length === 0 &&
-				config.getEnv('deployment.type') === 'internal'
-			) {
-				await this.executionRepository.setRunning(executionId); // set `startedAt` so we display it correctly in UI
-				throw new ApplicationError('Execution to enqueue has empty node execution stack', {
-					extra: { executionData: data.executionData },
-				});
-			}
-
 			job = await this.scalingService.addJob(jobData, { priority: realtime ? 50 : 100 });
 
 			hooks = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerMain(