diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index 2d03dac507..cbabd68504 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -202,7 +202,7 @@ describe('ScalingService', () => { }); describe('if worker', () => { - it('should wait for running jobs to finish', async () => { + it('should pause queue and wait for running jobs to finish', async () => { // @ts-expect-error readonly property instanceSettings.instanceType = 'worker'; await scalingService.setupQueue(); @@ -211,7 +211,7 @@ describe('ScalingService', () => { await scalingService.stop(); expect(getRunningJobsCountSpy).toHaveBeenCalled(); - expect(queue.pause).not.toHaveBeenCalled(); + expect(queue.pause).toHaveBeenCalled(); expect(stopQueueRecoverySpy).not.toHaveBeenCalled(); }); }); diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 231d607acd..ea9d2fe447 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -139,17 +139,21 @@ export class ScalingService { else if (instanceType === 'worker') await this.stopWorker(); } + private async pauseQueue() { + await this.queue.pause(true, true); // no more jobs will be enqueued or picked up + this.logger.debug('Paused queue'); + } + private async stopMain() { - if (this.instanceSettings.isSingleMain) { - await this.queue.pause(true, true); // no more jobs will be picked up - this.logger.debug('Queue paused'); - } + if (this.instanceSettings.isSingleMain) await this.pauseQueue(); if (this.queueRecoveryContext.timeout) this.stopQueueRecovery(); if (this.isQueueMetricsEnabled) this.stopQueueMetrics(); } private async stopWorker() { + await this.pauseQueue(); + let count = 0; while (this.getRunningJobsCount() !== 0) {