fix(core): Fix race condition when stopping jobs in queue mode (#27211)

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

@@ -5,7 +5,7 @@ import { Container } from '@n8n/di';
 import * as BullModule from 'bull';
 import { mock } from 'jest-mock-extended';
 import { InstanceSettings } from 'n8n-core';
-import { ApplicationError, ManualExecutionCancelledError } from 'n8n-workflow';
+import { ApplicationError } from 'n8n-workflow';
 
 import type { ActiveExecutions } from '@/active-executions';
 
@@ -288,15 +288,15 @@ describe('ScalingService', () => {
 	});
 
 	describe('stopJob', () => {
-		it('should stop an active job', async () => {
+		it('should stop an active job by sending abort signal only', async () => {
 			await scalingService.setupQueue();
 			const job = mock<Job>({ isActive: jest.fn().mockResolvedValue(true) });
 
 			const result = await scalingService.stopJob(job);
 
 			expect(job.progress).toHaveBeenCalledWith({ kind: 'abort-job' });
-			expect(job.discard).toHaveBeenCalled();
-			expect(job.moveToFailed).toHaveBeenCalledWith(new ManualExecutionCancelledError('123'), true);
+			expect(job.discard).not.toHaveBeenCalled();
+			expect(job.moveToFailed).not.toHaveBeenCalled();
 			expect(result).toBe(true);
 		});
 

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

@@ -23,6 +23,7 @@ import type {
 } from 'n8n-workflow';
 import {
 	BINARY_ENCODING,
+	ManualExecutionCancelledError,
 	NodeConnectionTypes,
 	Workflow,
 	UnexpectedError,
@@ -293,6 +294,10 @@ export class JobProcessor {
 
 		delete this.runningJobs[job.id];
 
+		if (run?.status === 'canceled') {
+			throw new ManualExecutionCancelledError(executionId);
+		}
+
 		const props = process.env.N8N_MINIMIZE_EXECUTION_DATA_FETCHING
 			? this.deriveJobFinishedProps(run, startedAt)
 			: await this.fetchJobFinishedResult(executionId);

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

@@ -5,14 +5,7 @@ import { ExecutionRepository } from '@n8n/db';
 import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Container, Service } from '@n8n/di';
 import { ErrorReporter, InstanceSettings } from 'n8n-core';
-import {
-	BINARY_ENCODING,
-	sleep,
-	jsonStringify,
-	ensureError,
-	UnexpectedError,
-	ManualExecutionCancelledError,
-} from 'n8n-workflow';
+import { BINARY_ENCODING, sleep, jsonStringify, ensureError, UnexpectedError } from 'n8n-workflow';
 import type { IExecuteResponsePromiseData, IRun } from 'n8n-workflow';
 import assert, { strict } from 'node:assert';
 
@@ -264,8 +257,7 @@ export class ScalingService {
 		try {
 			if (await job.isActive()) {
 				await job.progress({ kind: 'abort-job' }); // being processed by worker
-				await job.discard(); // prevent retries
-				await job.moveToFailed(new ManualExecutionCancelledError(job.data.executionId), true); // remove from queue
+				this.logger.debug('Sent abort signal to worker', props);
 				return true;
 			}