feat(core): Coordinate manual workflow activation and deactivation in multi-main scenario

packages/cli/src/ActiveWorkflowRunner.ts

@@ -2,7 +2,7 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 
-import Container, { Service } from 'typedi';
+import { Service } from 'typedi';
 import { ActiveWorkflows, NodeExecuteFunctions } from 'n8n-core';
 
 import type {
@@ -64,8 +64,7 @@ import { In } from 'typeorm';
 import { WebhookService } from './services/webhook.service';
 import { Logger } from './Logger';
 import { WorkflowRepository } from '@/databases/repositories';
-import config from '@/config';
-import type { MultiMainInstancePublisher } from './services/orchestration/main/MultiMainInstance.publisher.ee';
+import { MultiMainSetup } from './services/orchestration/main/MultiMainSetup.ee';
 
 const WEBHOOK_PROD_UNREGISTERED_HINT =
 	"The workflow must be active for a production URL to run successfully. You can activate the workflow using the toggle in the top-right of the editor. Note that unlike test URL calls, production URL calls aren't shown on the canvas (only in the executions list)";
@@ -92,31 +91,18 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 		};
 	} = {};
 
-	isMultiMainScenario =
-		config.getEnv('executions.mode') === 'queue' && config.getEnv('leaderSelection.enabled');
-
-	multiMainInstancePublisher: MultiMainInstancePublisher | undefined;
-
 	constructor(
 		private readonly logger: Logger,
 		private readonly activeExecutions: ActiveExecutions,
 		private readonly externalHooks: ExternalHooks,
 		private readonly nodeTypes: NodeTypes,
 		private readonly webhookService: WebhookService,
 		private readonly workflowRepository: WorkflowRepository,
+		private readonly multiMainSetup: MultiMainSetup,
 	) {}
 
 	async init() {
-		if (this.isMultiMainScenario) {
-			const { MultiMainInstancePublisher } = await import(
-				'@/services/orchestration/main/MultiMainInstance.publisher.ee'
-			);
-
-			this.multiMainInstancePublisher = Container.get(MultiMainInstancePublisher);
-
-			await this.multiMainInstancePublisher.init();
-		}
-
+		await this.multiMainSetup.init();
 		await this.addActiveWorkflows('init');
 
 		await this.externalHooks.run('activeWorkflows.initialized', []);
@@ -744,17 +730,34 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 		activationMode: WorkflowActivateMode,
 		existingWorkflow?: WorkflowEntity,
 	) {
+		if (this.multiMainSetup.isEnabled && this.multiMainSetup.isFollower) {
+			const prefix = '[Multi-main setup] Instance is follower';
+
+			this.logger.debug(
+				[prefix, 'skipping addition of workflow to active workflows in memory...'].join(', '),
+			);
+
+			if (['activate', 'update'].includes(activationMode)) {
+				this.logger.debug(
+					[prefix, 'broadcasting "workflowWasUpdated" into command channel...'].join(', '),
+				);
+				await this.multiMainSetup.broadcastWorkflowWasUpdated(workflowId);
+			}
+
+			return;
+		}
+
 		let workflow: Workflow;
 
 		let shouldAddWebhooks = true;
 		let shouldAddTriggersAndPollers = true;
 
-		if (this.isMultiMainScenario && activationMode !== 'leadershipChange') {
-			shouldAddWebhooks = this.multiMainInstancePublisher?.isLeader ?? false;
-			shouldAddTriggersAndPollers = this.multiMainInstancePublisher?.isLeader ?? false;
+		if (this.multiMainSetup.isEnabled && activationMode !== 'leadershipChange') {
+			shouldAddWebhooks = this.multiMainSetup.isLeader;
+			shouldAddTriggersAndPollers = this.multiMainSetup.isLeader;
 		}
 
-		if (this.isMultiMainScenario && activationMode === 'leadershipChange') {
+		if (this.multiMainSetup.isEnabled && activationMode === 'leadershipChange') {
 			shouldAddWebhooks = false;
 			shouldAddTriggersAndPollers = true;
 		}
@@ -940,6 +943,17 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	// TODO: this should happen in a transaction
 	async remove(workflowId: string) {
 		// Remove all the webhooks of the workflow
+
+		if (this.multiMainSetup.isEnabled && this.multiMainSetup.isFollower) {
+			this.logger.debug(
+				'[Multi-main setup] Instance is follower, skipping removal of workflow from active workflows in memory and broadcasting "workflowWasDeactivated" into command channel...',
+			);
+
+			await this.multiMainSetup.broadcastWorkflowWasDeactivated(workflowId);
+
+			return;
+		}
+
 		try {
 			await this.clearWebhooks(workflowId);
 		} catch (error) {

packages/cli/src/ExternalSecrets/ExternalSecretsManager.ee.ts

@@ -19,7 +19,7 @@ import {
 import { License } from '@/License';
 import { InternalHooks } from '@/InternalHooks';
 import { ExternalSecretsProviders } from './ExternalSecretsProviders.ee';
-import { SingleMainInstancePublisher } from '@/services/orchestration/main/SingleMainInstance.publisher';
+import { SingleMainSetup } from '@/services/orchestration/main/SingleMainSetup';
 
 @Service()
 export class ExternalSecretsManager {
@@ -82,7 +82,7 @@ export class ExternalSecretsManager {
 	}
 
 	async broadcastReloadExternalSecretsProviders() {
-		await Container.get(SingleMainInstancePublisher).broadcastReloadExternalSecretsProviders();
+		await Container.get(SingleMainSetup).broadcastReloadExternalSecretsProviders();
 	}
 
 	private decryptSecretsSettings(value: string): ExternalSecretsSettings {

packages/cli/src/Interfaces.ts

@@ -504,7 +504,19 @@ export type IPushData =
 	| PushDataTestWebhook
 	| PushDataNodeDescriptionUpdated
 	| PushDataExecutionRecovered
-	| PushDataActiveWorkflowUsersChanged;
+	| PushDataActiveWorkflowUsersChanged
+	| PushDataActiveWorkflowAdded
+	| PushDataActiveWorkflowRemoved;
+
+type PushDataActiveWorkflowAdded = {
+	data: IActiveWorkflowAdded;
+	type: 'workflowActivated';
+};
+
+type PushDataActiveWorkflowRemoved = {
+	data: IActiveWorkflowRemoved;
+	type: 'workflowDeactivated';
+};
 
 type PushDataActiveWorkflowUsersChanged = {
 	data: IActiveWorkflowUsersChanged;
@@ -566,6 +578,12 @@ export interface IActiveWorkflowUser {
 	lastSeen: Date;
 }
 
+export interface IActiveWorkflowAdded {
+	workflowId: Workflow['id'];
+}
+
+type IActiveWorkflowRemoved = IActiveWorkflowAdded;
+
 export interface IActiveWorkflowUsersChanged {
 	workflowId: Workflow['id'];
 	activeUsers: IActiveWorkflowUser[];

packages/cli/src/License.ts

@@ -16,6 +16,7 @@ import { WorkflowRepository } from '@/databases/repositories';
 import type { BooleanLicenseFeature, N8nInstanceType, NumericLicenseFeature } from './Interfaces';
 import type { RedisServicePubSubPublisher } from './services/redis/RedisServicePubSubPublisher';
 import { RedisService } from './services/redis.service';
+import { MultiMainSetup } from '@/services/orchestration/main/MultiMainSetup.ee';
 
 type FeatureReturnType = Partial<
 	{
@@ -40,13 +41,16 @@ export class License {
 	constructor(
 		private readonly logger: Logger,
 		private readonly instanceSettings: InstanceSettings,
+		private readonly multiMainSetup: MultiMainSetup,
 	) {}
 
 	async init(instanceType: N8nInstanceType = 'main') {
 		if (this.manager) {
 			return;
 		}
 
+		await this.multiMainSetup.init();
+
 		const isMainInstance = instanceType === 'main';
 		const server = config.getEnv('license.serverUrl');
 		const autoRenewEnabled = isMainInstance && config.getEnv('license.autoRenewEnabled');
@@ -112,22 +116,14 @@ export class License {
 	}
 
 	async onFeatureChange(_features: TFeatures): Promise<void> {
-		if (config.getEnv('executions.mode') === 'queue') {
-			if (config.getEnv('leaderSelection.enabled')) {
-				const { MultiMainInstancePublisher } = await import(
-					'@/services/orchestration/main/MultiMainInstance.publisher.ee'
-				);
-
-				const multiMainInstancePublisher = Container.get(MultiMainInstancePublisher);
-
-				await multiMainInstancePublisher.init();
-
-				if (multiMainInstancePublisher.isFollower) {
-					this.logger.debug('Instance is follower, skipping sending of reloadLicense command...');
-					return;
-				}
-			}
+		if (this.multiMainSetup.isEnabled && this.multiMainSetup.isFollower) {
+			this.logger.debug(
+				'[Multi-main setup] Instance is follower, skipping sending of "reloadLicense" command...',
+			);
+			return;
+		}
 
+		if (config.getEnv('executions.mode') === 'queue') {
 			if (!this.redisPublisher) {
 				this.logger.debug('Initializing Redis publisher for License Service');
 				this.redisPublisher = await Container.get(RedisService).getPubSubPublisher();

packages/cli/src/commands/BaseCommand.ts

@@ -22,6 +22,7 @@ import { ExternalSecretsManager } from '@/ExternalSecrets/ExternalSecretsManager
 import { initExpressionEvaluator } from '@/ExpressionEvalator';
 import { generateHostInstanceId } from '../databases/utils/generators';
 import { WorkflowHistoryManager } from '@/workflows/workflowHistory/workflowHistoryManager.ee';
+import { MultiMainSetup } from '@/services/orchestration/main/MultiMainSetup.ee';
 
 export abstract class BaseCommand extends Command {
 	protected logger = Container.get(Logger);
@@ -243,19 +244,16 @@ export abstract class BaseCommand extends Command {
 	}
 
 	async initLicense(): Promise<void> {
-		if (config.getEnv('executions.mode') === 'queue' && config.getEnv('leaderSelection.enabled')) {
-			const { MultiMainInstancePublisher } = await import(
-				'@/services/orchestration/main/MultiMainInstance.publisher.ee'
-			);
+		const multiMainSetup = Container.get(MultiMainSetup);
 
-			const multiMainInstancePublisher = Container.get(MultiMainInstancePublisher);
+		await multiMainSetup.init();
 
-			await multiMainInstancePublisher.init();
+		if (multiMainSetup.isEnabled && multiMainSetup.isFollower) {
+			this.logger.debug(
+				'[Multi-main setup] Instance is follower, skipping license initialization...',
+			);
 
-			if (multiMainInstancePublisher.isFollower) {
-				this.logger.debug('Instance is follower, skipping license initialization...');
-				return;
-			}
+			return;
 		}
 
 		const license = Container.get(License);

packages/cli/src/commands/start.ts

@@ -27,9 +27,10 @@ import { BaseCommand } from './BaseCommand';
 import { InternalHooks } from '@/InternalHooks';
 import { License, FeatureNotLicensedError } from '@/License';
 import { IConfig } from '@oclif/config';
-import { SingleMainInstancePublisher } from '@/services/orchestration/main/SingleMainInstance.publisher';
+import { SingleMainSetup } from '@/services/orchestration/main/SingleMainSetup';
 import { OrchestrationHandlerMainService } from '@/services/orchestration/main/orchestration.handler.main.service';
 import { PruningService } from '@/services/pruning.service';
+import { MultiMainSetup } from '@/services/orchestration/main/MultiMainSetup.ee';
 
 // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-var-requires
 const open = require('open');
@@ -116,14 +117,12 @@ export class Start extends BaseCommand {
 				await this.pruningService.stopPruning();
 			}
 
-			if (config.getEnv('leaderSelection.enabled')) {
-				const { MultiMainInstancePublisher } = await import(
-					'@/services/orchestration/main/MultiMainInstance.publisher.ee'
-				);
+			const multiMainSetup = Container.get(MultiMainSetup);
 
+			if (multiMainSetup.isEnabled) {
 				await this.activeWorkflowRunner.removeAllTriggerAndPollerBasedWorkflows();
 
-				await Container.get(MultiMainInstancePublisher).destroy();
+				await multiMainSetup.shutdown();
 			}
 
 			await Container.get(InternalHooks).onN8nStop();
@@ -233,32 +232,25 @@ export class Start extends BaseCommand {
 		if (config.get('executions.mode') !== 'queue') return;
 
 		if (!config.get('leaderSelection.enabled')) {
-			await Container.get(SingleMainInstancePublisher).init();
+			await Container.get(SingleMainSetup).init();
 			await Container.get(OrchestrationHandlerMainService).init();
 			return;
 		}
 
 		// multi-main scenario
 
-		const { MultiMainInstancePublisher } = await import(
-			'@/services/orchestration/main/MultiMainInstance.publisher.ee'
-		);
+		const multiMainSetup = Container.get(MultiMainSetup);
 
-		const multiMainInstancePublisher = Container.get(MultiMainInstancePublisher);
+		await multiMainSetup.init();
 
-		await multiMainInstancePublisher.init();
-
-		if (
-			multiMainInstancePublisher.isLeader &&
-			!Container.get(License).isMultipleMainInstancesLicensed()
-		) {
+		if (multiMainSetup.isLeader && !Container.get(License).isMultipleMainInstancesLicensed()) {
 			throw new FeatureNotLicensedError(LICENSE_FEATURES.MULTIPLE_MAIN_INSTANCES);
 		}
 
 		await Container.get(OrchestrationHandlerMainService).init();
 
-		multiMainInstancePublisher.on('leadershipChange', async () => {
-			if (multiMainInstancePublisher.isLeader) {
+		multiMainSetup.on('leadershipChange', async () => {
+			if (multiMainSetup.isLeader) {
 				await this.activeWorkflowRunner.addAllTriggerAndPollerBasedWorkflows();
 			} else {
 				// only in case of leadership change without shutdown

packages/cli/src/controllers/orchestration.controller.ts

@@ -1,13 +1,13 @@
 import { Authorized, Get, RestController } from '@/decorators';
 import { OrchestrationRequest } from '@/requests';
 import { Service } from 'typedi';
-import { SingleMainInstancePublisher } from '@/services/orchestration/main/SingleMainInstance.publisher';
+import { SingleMainSetup } from '@/services/orchestration/main/SingleMainSetup';
 
 @Authorized(['global', 'owner'])
 @RestController('/orchestration')
 @Service()
 export class OrchestrationController {
-	constructor(private readonly orchestrationService: SingleMainInstancePublisher) {}
+	constructor(private readonly orchestrationService: SingleMainSetup) {}
 
 	/**
 	 * These endpoint currently do not return anything, they just trigger the messsage to

packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts

@@ -32,7 +32,7 @@ import { Container, Service } from 'typedi';
 import { ExecutionRepository, WorkflowRepository } from '@/databases/repositories';
 import type { AbstractEventMessageOptions } from '../EventMessageClasses/AbstractEventMessageOptions';
 import { getEventMessageObjectByType } from '../EventMessageClasses/Helpers';
-import { SingleMainInstancePublisher } from '@/services/orchestration/main/SingleMainInstance.publisher';
+import { SingleMainSetup } from '@/services/orchestration/main/SingleMainSetup';
 import { Logger } from '@/Logger';
 
 export type EventMessageReturnMode = 'sent' | 'unsent' | 'all' | 'unfinished';
@@ -206,9 +206,7 @@ export class MessageEventBus extends EventEmitter {
 		this.destinations[destination.getId()] = destination;
 		this.destinations[destination.getId()].startListening();
 		if (notifyWorkers) {
-			await Container.get(
-				SingleMainInstancePublisher,
-			).broadcastRestartEventbusAfterDestinationUpdate();
+			await Container.get(SingleMainSetup).broadcastRestartEventbusAfterDestinationUpdate();
 		}
 		return destination;
 	}
@@ -234,9 +232,7 @@ export class MessageEventBus extends EventEmitter {
 			delete this.destinations[id];
 		}
 		if (notifyWorkers) {
-			await Container.get(
-				SingleMainInstancePublisher,
-			).broadcastRestartEventbusAfterDestinationUpdate();
+			await Container.get(SingleMainSetup).broadcastRestartEventbusAfterDestinationUpdate();
 		}
 		return result;
 	}

packages/cli/src/services/orchestration.base.service.ts

@@ -5,7 +5,7 @@ import config from '@/config';
 import { EventEmitter } from 'node:events';
 
 export abstract class OrchestrationService extends EventEmitter {
-	protected initialized = false;
+	protected isInitialized = false;
 
 	protected queueModeId: string;
 
@@ -36,17 +36,17 @@ export abstract class OrchestrationService extends EventEmitter {
 	}
 
 	sanityCheck(): boolean {
-		return this.initialized && this.isQueueMode;
+		return this.isInitialized && this.isQueueMode;
 	}
 
 	async init() {
 		await this.initPublisher();
-		this.initialized = true;
+		this.isInitialized = true;
 	}
 
 	async shutdown() {
 		await this.redisPublisher?.destroy();
-		this.initialized = false;
+		this.isInitialized = false;
 	}
 
 	protected async initPublisher() {