fix(core): Make senderId required for all command messages (#7252)

all commands sent between main instance and workers need to contain a
server id to prevent senders from reacting to their own messages,
causing loops

this PR makes sure all sent messages contain a sender id by default as
part of constructing a sending redis client.

---------

Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <aditya@netroy.in>

packages/cli/src/commands/BaseCommand.ts

@@ -22,6 +22,7 @@ import { PostHogClient } from '@/posthog';
 import { License } from '@/License';
 import { ExternalSecretsManager } from '@/ExternalSecrets/ExternalSecretsManager.ee';
 import { initExpressionEvaluator } from '@/ExpressionEvalator';
+import { generateHostInstanceId } from '../databases/utils/generators';
 
 export abstract class BaseCommand extends Command {
 	protected logger = LoggerProxy.init(getLogger());
@@ -36,6 +37,10 @@ export abstract class BaseCommand extends Command {
 
 	protected instanceId: string;
 
+	instanceType: N8nInstanceType = 'main';
+
+	queueModeId: string;
+
 	protected server?: AbstractServer;
 
 	async init(): Promise<void> {
@@ -83,6 +88,22 @@ export abstract class BaseCommand extends Command {
 		await Container.get(InternalHooks).init(this.instanceId);
 	}
 
+	protected setInstanceType(instanceType: N8nInstanceType) {
+		this.instanceType = instanceType;
+		config.set('generic.instanceType', instanceType);
+	}
+
+	protected setInstanceQueueModeId() {
+		if (config.getEnv('executions.mode') === 'queue') {
+			if (config.get('redis.queueModeId')) {
+				this.queueModeId = config.get('redis.queueModeId');
+				return;
+			}
+			this.queueModeId = generateHostInstanceId(this.instanceType);
+			config.set('redis.queueModeId', this.queueModeId);
+		}
+	}
+
 	protected async stopProcess() {
 		// This needs to be overridden
 	}
@@ -115,11 +136,9 @@ export abstract class BaseCommand extends Command {
 		await this.externalHooks.init();
 	}
 
-	async initLicense(instanceType: N8nInstanceType = 'main'): Promise<void> {
-		config.set('generic.instanceType', instanceType);
-
+	async initLicense(): Promise<void> {
 		const license = Container.get(License);
-		await license.init(this.instanceId, instanceType);
+		await license.init(this.instanceId, this.instanceType ?? 'main');
 
 		const activationKey = config.getEnv('license.activationKey');
 

packages/cli/src/commands/start.ts

@@ -30,6 +30,7 @@ import { BaseCommand } from './BaseCommand';
 import { InternalHooks } from '@/InternalHooks';
 import { License } from '@/License';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
+import { IConfig } from '@oclif/config';
 
 // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-var-requires
 const open = require('open');
@@ -65,6 +66,12 @@ export class Start extends BaseCommand {
 
 	protected server = new Server();
 
+	constructor(argv: string[], cmdConfig: IConfig) {
+		super(argv, cmdConfig);
+		this.setInstanceType('main');
+		this.setInstanceQueueModeId();
+	}
+
 	/**
 	 * Opens the UI in browser
 	 */
@@ -196,11 +203,16 @@ export class Start extends BaseCommand {
 	async init() {
 		await this.initCrashJournal();
 
-		await super.init();
 		this.logger.info('Initializing n8n process');
+		if (config.getEnv('executions.mode') === 'queue') {
+			this.logger.debug('Main Instance running in queue mode');
+			this.logger.debug(`Queue mode id: ${this.queueModeId}`);
+		}
+
+		await super.init();
 		this.activeWorkflowRunner = Container.get(ActiveWorkflowRunner);
 
-		await this.initLicense('main');
+		await this.initLicense();
 		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();

packages/cli/src/commands/webhook.ts

@@ -6,6 +6,7 @@ import { WebhookServer } from '@/WebhookServer';
 import { Queue } from '@/Queue';
 import { BaseCommand } from './BaseCommand';
 import { Container } from 'typedi';
+import { IConfig } from '@oclif/config';
 
 export class Webhook extends BaseCommand {
 	static description = 'Starts n8n webhook process. Intercepts only production URLs.';
@@ -18,6 +19,15 @@ export class Webhook extends BaseCommand {
 
 	protected server = new WebhookServer();
 
+	constructor(argv: string[], cmdConfig: IConfig) {
+		super(argv, cmdConfig);
+		this.setInstanceType('webhook');
+		if (this.queueModeId) {
+			this.logger.debug(`Webhook Instance queue mode id: ${this.queueModeId}`);
+		}
+		this.setInstanceQueueModeId();
+	}
+
 	/**
 	 * Stops n8n in a graceful way.
 	 * Make for example sure that all the webhooks from third party services
@@ -75,9 +85,13 @@ export class Webhook extends BaseCommand {
 		}
 
 		await this.initCrashJournal();
+
+		this.logger.info('Initializing n8n webhook process');
+		this.logger.debug(`Queue mode id: ${this.queueModeId}`);
+
 		await super.init();
 
-		await this.initLicense('webhook');
+		await this.initLicense();
 		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();

packages/cli/src/commands/worker.ts

@@ -29,7 +29,6 @@ import { N8N_VERSION } from '@/constants';
 import { BaseCommand } from './BaseCommand';
 import { ExecutionRepository } from '@db/repositories';
 import { OwnershipService } from '@/services/ownership.service';
-import { generateHostInstanceId } from '@/databases/utils/generators';
 import type { ICredentialsOverwrite } from '@/Interfaces';
 import { CredentialsOverwrites } from '@/CredentialsOverwrites';
 import { rawBodyReader, bodyParser } from '@/middlewares';
@@ -38,6 +37,7 @@ import { RedisServicePubSubPublisher } from '../services/redis/RedisServicePubSu
 import { RedisServicePubSubSubscriber } from '../services/redis/RedisServicePubSubSubscriber';
 import { EventMessageGeneric } from '../eventbus/EventMessageClasses/EventMessageGeneric';
 import { getWorkerCommandReceivedHandler } from '../worker/workerCommandHandler';
+import { IConfig } from '@oclif/config';
 
 export class Worker extends BaseCommand {
 	static description = '\nStarts a n8n worker';
@@ -58,8 +58,6 @@ export class Worker extends BaseCommand {
 
 	static jobQueue: JobQueue;
 
-	readonly uniqueInstanceId = generateHostInstanceId('worker');
-
 	redisPublisher: RedisServicePubSubPublisher;
 
 	redisSubscriber: RedisServicePubSubSubscriber;
@@ -250,13 +248,22 @@ export class Worker extends BaseCommand {
 		};
 	}
 
+	constructor(argv: string[], cmdConfig: IConfig) {
+		super(argv, cmdConfig);
+		this.setInstanceType('worker');
+		this.setInstanceQueueModeId();
+	}
+
 	async init() {
 		await this.initCrashJournal();
-		await super.init();
-		this.logger.debug(`Worker ID: ${this.uniqueInstanceId}`);
-		this.logger.debug('Starting n8n worker...');
 
-		await this.initLicense('worker');
+		this.logger.debug('Starting n8n worker...');
+		this.logger.debug(`Queue mode id: ${this.queueModeId}`);
+
+		await super.init();
+
+		await this.initLicense();
+
 		await this.initBinaryDataService();
 		await this.initExternalHooks();
 		await this.initExternalSecrets();
@@ -267,8 +274,7 @@ export class Worker extends BaseCommand {
 
 	async initEventBus() {
 		await eventBus.initialize({
-			workerId: this.uniqueInstanceId,
-			uniqueInstanceId: this.uniqueInstanceId,
+			workerId: this.queueModeId,
 		});
 	}
 
@@ -286,7 +292,7 @@ export class Worker extends BaseCommand {
 			new EventMessageGeneric({
 				eventName: 'n8n.worker.started',
 				payload: {
-					workerId: this.uniqueInstanceId,
+					workerId: this.queueModeId,
 				},
 			}),
 		);
@@ -295,7 +301,7 @@ export class Worker extends BaseCommand {
 			'WorkerCommandReceivedHandler',
 			// eslint-disable-next-line @typescript-eslint/no-unsafe-argument
 			getWorkerCommandReceivedHandler({
-				uniqueInstanceId: this.uniqueInstanceId,
+				queueModeId: this.queueModeId,
 				instanceId: this.instanceId,
 				redisPublisher: this.redisPublisher,
 				getRunningJobIds: () => Object.keys(Worker.runningJobs),