fix(core): Fix worker shutdown errors when active executions (#10353)

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

@@ -7,6 +7,7 @@ import type { Job, JobData, JobOptions, JobQueue } from '../types';
 import { ApplicationError } from 'n8n-workflow';
 import { mockInstance } from '@test/mocking';
 import { GlobalConfig } from '@n8n/config';
+import type { JobProcessor } from '../job-processor';
 
 const queue = mock<JobQueue>({
 	client: { ping: jest.fn() },
@@ -100,23 +101,27 @@ describe('ScalingService', () => {
 		});
 	});
 
-	describe('pauseQueue', () => {
-		it('should pause the queue', async () => {
+	describe('stop', () => {
+		it('should pause the queue and check for running jobs', async () => {
 			/**
 			 * Arrange
 			 */
-			const scalingService = new ScalingService(mock(), mock(), mock(), globalConfig);
+			const jobProcessor = mock<JobProcessor>();
+			const scalingService = new ScalingService(mock(), mock(), jobProcessor, globalConfig);
 			await scalingService.setupQueue();
+			jobProcessor.getRunningJobIds.mockReturnValue([]);
+			const getRunningJobsCountSpy = jest.spyOn(scalingService, 'getRunningJobsCount');
 
 			/**
 			 * Act
 			 */
-			await scalingService.pauseQueue();
+			await scalingService.stop();
 
 			/**
 			 * Assert
 			 */
 			expect(queue.pause).toHaveBeenCalledWith(true, true);
+			expect(getRunningJobsCountSpy).toHaveBeenCalled();
 		});
 	});
 

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

@@ -1,5 +1,5 @@
 import Container, { Service } from 'typedi';
-import { ApplicationError, BINARY_ENCODING } from 'n8n-workflow';
+import { ApplicationError, BINARY_ENCODING, sleep } from 'n8n-workflow';
 import { ActiveExecutions } from '@/ActiveExecutions';
 import config from '@/config';
 import { Logger } from '@/Logger';
@@ -59,10 +59,22 @@ export class ScalingService {
 	}
 
 	@OnShutdown(HIGHEST_SHUTDOWN_PRIORITY)
-	async pauseQueue() {
+	async stop() {
 		await this.queue.pause(true, true);
 
 		this.logger.debug('[ScalingService] Queue paused');
+
+		let count = 0;
+
+		while (this.getRunningJobsCount() !== 0) {
+			if (count++ % 4 === 0) {
+				this.logger.info(
+					`Waiting for ${this.getRunningJobsCount()} active executions to finish...`,
+				);
+			}
+
+			await sleep(500);
+		}
 	}
 
 	async pingQueue() {
@@ -113,6 +125,10 @@ export class ScalingService {
 		}
 	}
 
+	getRunningJobsCount() {
+		return this.jobProcessor.getRunningJobIds().length;
+	}
+
 	// #endregion
 
 	// #region Listeners