refactor(core): Centralize scaling mode (no-changelog)

packages/cli/BREAKING-CHANGES.md

@@ -2,6 +2,16 @@
 
 This list shows all the versions which include breaking changes and how to upgrade.
 
+## 1.??.? @TODO: Update version
+
+### What changed?
+
+n8n upgraded its scaling mode from `bull@4.12.1` to `bullmq@5.8.2`.
+
+### When is action necessary?
+
+If you are using queue mode, before you upgrade, stop all running workflows, wait until all jobs in the queue have completed, stop all main instances, and stop all workers. Then upgrade the main instance and all workers to the new n8n version and restart them. Additionally, ensure you are running Redis version 5 or above, which supports Redis Streams.
+
 ## 1.47.0
 
 ### What changed?

packages/cli/package.json

@@ -109,7 +109,7 @@
     "axios": "1.6.7",
     "basic-auth": "2.0.1",
     "bcryptjs": "2.4.3",
-    "bull": "4.12.1",
+    "bullmq": "5.8.2",
     "cache-manager": "5.2.3",
     "callsites": "3.1.0",
     "change-case": "4.1.2",

packages/cli/src/Server.ts

@@ -210,8 +210,8 @@ export class Server extends AbstractServer {
 		setupPushHandler(restEndpoint, app);
 
 		if (config.getEnv('executions.mode') === 'queue') {
-			const { Queue } = await import('@/Queue');
-			await Container.get(Queue).init();
+			const { ScalingMode } = await import('@/scaling-mode/scaling-mode');
+			await Container.get(ScalingMode).setupQueue();
 		}
 
 		await handleMfaDisable();

packages/cli/src/WebhookHelpers.ts

@@ -20,7 +20,6 @@ import type {
 	IDataObject,
 	IDeferredPromise,
 	IExecuteData,
-	IExecuteResponsePromiseData,
 	IHttpRequestMethods,
 	IN8nHttpFullResponse,
 	INode,
@@ -192,19 +191,6 @@ export function getWorkflowWebhooks(
 	return returnData;
 }
 
-export function encodeWebhookResponse(
-	response: IExecuteResponsePromiseData,
-): IExecuteResponsePromiseData {
-	if (typeof response === 'object' && Buffer.isBuffer(response.body)) {
-		response.body = {
-			// eslint-disable-next-line @typescript-eslint/naming-convention
-			'__@N8nEncodedBuffer@__': response.body.toString(BINARY_ENCODING),
-		};
-	}
-
-	return response;
-}
-
 const normalizeFormData = <T>(values: Record<string, T | T[]>) => {
 	for (const key in values) {
 		const value = values[key];

packages/cli/src/WorkflowRunner.ts

@@ -28,8 +28,8 @@ import { ExecutionRepository } from '@db/repositories/execution.repository';
 import { ExternalHooks } from '@/ExternalHooks';
 import type { IExecutionResponse, IWorkflowExecutionDataProcess } from '@/Interfaces';
 import { NodeTypes } from '@/NodeTypes';
-import type { Job, JobData, JobResponse } from '@/Queue';
-import { Queue } from '@/Queue';
+import type { Job, JobData, JobResult } from './scaling-mode/types';
+import { ScalingMode } from '@/scaling-mode/scaling-mode';
 import * as WorkflowHelpers from '@/WorkflowHelpers';
 import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
 import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
@@ -41,7 +41,7 @@ import { EventRelay } from './eventbus/event-relay.service';
 
 @Service()
 export class WorkflowRunner {
-	private jobQueue: Queue;
+	private scalingMode: ScalingMode;
 
 	private executionsMode = config.getEnv('executions.mode');
 
@@ -56,7 +56,7 @@ export class WorkflowRunner {
 		private readonly eventRelay: EventRelay,
 	) {
 		if (this.executionsMode === 'queue') {
-			this.jobQueue = Container.get(Queue);
+			this.scalingMode = Container.get(ScalingMode);
 		}
 	}
 
@@ -384,9 +384,7 @@ export class WorkflowRunner {
 		let job: Job;
 		let hooks: WorkflowHooks;
 		try {
-			job = await this.jobQueue.add(jobData, jobOptions);
-
-			this.logger.info(`Started with job ID: ${job.id.toString()} (Execution ID: ${executionId})`);
+			job = await this.scalingMode.addJob(jobData, jobOptions);
 
 			hooks = WorkflowExecuteAdditionalData.getWorkflowHooksWorkerMain(
 				data.executionMode,
@@ -415,8 +413,7 @@ export class WorkflowRunner {
 			async (resolve, reject, onCancel) => {
 				onCancel.shouldReject = false;
 				onCancel(async () => {
-					const queue = Container.get(Queue);
-					await queue.stopJob(job);
+					await Container.get(ScalingMode).stopJob(job);
 
 					// We use "getWorkflowHooksWorkerExecuter" as "getWorkflowHooksWorkerMain" does not contain the
 					// "workflowExecuteAfter" which we require.
@@ -433,11 +430,17 @@ export class WorkflowRunner {
 					reject(error);
 				});
 
-				const jobData: Promise<JobResponse> = job.finished();
+				/**
+				 * @TODO: Replacing `job.finished()` with `job.waitUntilFinished()` will require `QueueEvents`,
+				 * which might be avoidable complexity. Do we need to preserve `watchDogInterval` in `bullmq`?
+				 */
+				// const jobData: Promise<JobResult> = job.finished();
+				// const jobData = job.waitUntilFinished(this.scalingMode.queueEvents);
 
 				const queueRecoveryInterval = config.getEnv('queue.bull.queueRecoveryInterval');
 
-				const racingPromises: Array<Promise<JobResponse>> = [jobData];
+				// const racingPromises: Array<Promise<JobResult>> = [jobData];
+				const racingPromises: Array<Promise<JobResult>> = [];
 
 				let clearWatchdogInterval;
 				if (queueRecoveryInterval > 0) {
@@ -455,9 +458,9 @@ export class WorkflowRunner {
 					 ************************************************ */
 					let watchDogInterval: NodeJS.Timeout | undefined;
 
-					const watchDog: Promise<JobResponse> = new Promise((res) => {
+					const watchDog: Promise<JobResult> = new Promise((res) => {
 						watchDogInterval = setInterval(async () => {
-							const currentJob = await this.jobQueue.getJob(job.id);
+							const currentJob = await this.scalingMode.getJob(job.id);
 							// When null means job is finished (not found in queue)
 							if (currentJob === null) {
 								// Mimic worker's success message
@@ -477,7 +480,7 @@ export class WorkflowRunner {
 				}
 
 				try {
-					await Promise.race(racingPromises);
+					// await Promise.race(racingPromises);
 					if (clearWatchdogInterval !== undefined) {
 						clearWatchdogInterval();
 					}

packages/cli/src/commands/webhook.ts

@@ -5,7 +5,7 @@ import { ApplicationError } from 'n8n-workflow';
 import config from '@/config';
 import { ActiveExecutions } from '@/ActiveExecutions';
 import { WebhookServer } from '@/WebhookServer';
-import { Queue } from '@/Queue';
+import { ScalingMode } from '@/scaling-mode/scaling-mode';
 import { BaseCommand } from './BaseCommand';
 
 import { OrchestrationWebhookService } from '@/services/orchestration/webhook/orchestration.webhook.service';
@@ -94,7 +94,7 @@ export class Webhook extends BaseCommand {
 			);
 		}
 
-		await Container.get(Queue).init();
+		await Container.get(ScalingMode).setupQueue();
 		await this.server.start();
 		this.logger.debug(`Webhook listener ID: ${this.server.uniqueInstanceId}`);
 		this.logger.info('Webhook listener waiting for requests.');