diff --git a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts index 3c4b311fe87..92565043b47 100644 --- a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts +++ b/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.findSingleExecution.mockResolvedValue( + mock({ + mode: 'manual', + workflowData: { nodes: [], staticData: {} }, + data: mock({ + executionData: undefined, + }), + }), + ); + + const manualExecutionService = mock(); + manualExecutionService.runManually.mockReturnValue( + Promise.reject(new Error('workflow run rejected')) as ReturnType< + ManualExecutionService['runManually'] + >, + ); + + const jobProcessor = new JobProcessor( + logger, + mock(), + executionPersistence, + mock(), + mock(), + mock(), + manualExecutionService, + executionsConfig, + mock(), + mock(), + ); + + const job = mock({ 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(); const executionPersistence = mock(); diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index bb5ec7e7433..2bdd5bb09f4 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/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'] }, + ); + }); }); }); diff --git a/packages/cli/src/scaling/job-processor.ts b/packages/cli/src/scaling/job-processor.ts index 4857a1da276..b12fb6898ea 100644 --- a/packages/cli/src/scaling/job-processor.ts +++ b/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); diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 878e8c35325..536398d6ee2 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/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 }, ); }