fix(core): Ensure running job cleanup when a workflow run rejects (#37621)

Co-authored-by: Claude Fable 5 <[email protected]>

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

@@ -217,6 +217,45 @@ describe('JobProcessor', () => {
 		},
 	);
 
+	it('should remove the running job entry when the workflow run rejects', async () => {
+		const executionPersistence = mock<ExecutionPersistence>();
+		executionPersistence.findSingleExecution.mockResolvedValue(
+			mock<IExecutionResponse>({
+				mode: 'manual',
+				workflowData: { nodes: [], staticData: {} },
+				data: mock<IRunExecutionData>({
+					executionData: undefined,
+				}),
+			}),
+		);
+
+		const manualExecutionService = mock<ManualExecutionService>();
+		manualExecutionService.runManually.mockReturnValue(
+			Promise.reject(new Error('workflow run rejected')) as ReturnType<
+				ManualExecutionService['runManually']
+			>,
+		);
+
+		const jobProcessor = new JobProcessor(
+			logger,
+			mock<ExecutionRepository>(),
+			executionPersistence,
+			mock(),
+			mock(),
+			mock(),
+			manualExecutionService,
+			executionsConfig,
+			mock(),
+			mock(),
+		);
+
+		const job = mock<Job>({ id: 'job-1' });
+
+		await expect(jobProcessor.processJob(job)).rejects.toThrow('workflow run rejected');
+
+		expect(jobProcessor.getRunningJobIds()).toEqual([]);
+	});
+
 	it('should send job-finished with success=false when execution has errors', async () => {
 		const executionRepository = mock<ExecutionRepository>();
 		const executionPersistence = mock<ExecutionPersistence>();

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

@@ -258,6 +258,22 @@ describe('ScalingService', () => {
 				expect(queue.pause).toHaveBeenCalled();
 				expect(stopQueueRecoverySpy).not.toHaveBeenCalled();
 			});
+
+			it('should log the execution IDs it is waiting for while draining', async () => {
+				// @ts-expect-error readonly property
+				instanceSettings.instanceType = 'worker';
+				await scalingService.setupQueue();
+				jobProcessor.getRunningJobIds.mockReturnValueOnce(['1']).mockReturnValue([]);
+				jobProcessor.getRunningJobsSummary.mockReturnValue([mock({ executionId: 'exec-1' })]);
+
+				await scalingService.stop();
+
+				// @ts-expect-error private property
+				expect(scalingService.logger.info).toHaveBeenCalledWith(
+					'Waiting for 1 active executions to finish... (execution IDs: exec-1)',
+					{ executionIds: ['exec-1'] },
+				);
+			});
 		});
 	});
 

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

@@ -329,9 +329,14 @@ export class JobProcessor {
 
 		this.runningJobs[job.id] = runningJob;
 
-		const run = await workflowRun;
-
-		delete this.runningJobs[job.id];
+		let run: IRun;
+		try {
+			run = await workflowRun;
+		} finally {
+			// An entry left behind on rejection keeps the count of running jobs
+			// above zero forever, which prevents shutdown from ever completing.
+			delete this.runningJobs[job.id];
+		}
 
 		if (run?.status === 'canceled') {
 			throw new ManualExecutionCancelledError(executionId);

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

@@ -186,8 +186,11 @@ export class ScalingService {
 
 		while (this.getRunningJobsCount() !== 0) {
 			if (count++ % 4 === 0) {
+				const summaries = this.jobProcessor.getRunningJobsSummary();
+				const executionIds = summaries.map((summary) => summary.executionId);
 				this.logger.info(
-					`Waiting for ${this.getRunningJobsCount()} active executions to finish...`,
+					`Waiting for ${executionIds.length} active executions to finish... (execution IDs: ${executionIds.join(', ')})`,
+					{ executionIds },
 				);
 			}