refactor(core): Streamline flows in multi-main mode (no-changelog) (#8446)

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

@@ -6,6 +6,7 @@ import type { RedisServiceBaseCommand, RedisServiceCommand } from './redis/Redis
 
 import { RedisService } from './redis.service';
 import { MultiMainSetup } from './orchestration/main/MultiMainSetup.ee';
+import type { WorkflowActivateMode } from 'n8n-workflow';
 
 @Service()
 export class OrchestrationService {
@@ -118,4 +119,29 @@ export class OrchestrationService {
 
 		await this.redisPublisher.publishToCommandChannel({ command });
 	}
+
+	// ----------------------------------
+	//           activations
+	// ----------------------------------
+
+	/**
+	 * Whether this instance may add webhooks to the `webhook_entity` table.
+	 */
+	shouldAddWebhooks(activationMode: WorkflowActivateMode) {
+		if (activationMode === 'init') return false;
+
+		if (activationMode === 'leadershipChange') return false;
+
+		return this.isLeader; // 'update' or 'activate'
+	}
+
+	/**
+	 * Whether this instance may add triggers and pollers to memory.
+	 *
+	 * In both single- and multi-main setup, only the leader is allowed to manage
+	 * triggers and pollers in memory, to ensure they are not duplicated.
+	 */
+	shouldAddTriggersAndPollers() {
+		return this.isLeader;
+	}
 }

packages/cli/src/services/orchestration/main/MultiMainSetup.ee.ts

@@ -62,11 +62,9 @@ export class MultiMainSetup extends EventEmitter {
 			if (config.getEnv('multiMainSetup.instanceType') === 'leader') {
 				config.set('multiMainSetup.instanceType', 'follower');
 
-				this.emit('leadershipChange'); // stop triggers, pollers, pruning
+				this.emit('leader-stepdown'); // lost leadership - stop triggers, pollers, pruning
 
-				EventReporter.report('[Multi-main setup] Leader failed to renew leader key', {
-					level: 'info',
-				});
+				EventReporter.info('[Multi-main setup] Leader failed to renew leader key');
 			}
 
 			return;
@@ -79,7 +77,7 @@ export class MultiMainSetup extends EventEmitter {
 
 			config.set('multiMainSetup.instanceType', 'follower');
 
-			this.emit('leadershipVacant'); // stop triggers, pollers, pruning
+			this.emit('leader-stepdown'); // lost leadership - stop triggers, pollers, pruning
 
 			await this.tryBecomeLeader();
 		}
@@ -99,7 +97,7 @@ export class MultiMainSetup extends EventEmitter {
 
 			await this.redisPublisher.setExpiration(this.leaderKey, this.leaderKeyTtl);
 
-			this.emit('leadershipChange'); // start triggers, pollers, pruning
+			this.emit('leader-takeover'); // gained leadership - start triggers, pollers, pruning
 		} else {
 			config.set('multiMainSetup.instanceType', 'follower');
 		}

packages/cli/src/services/orchestration/main/handleCommandMessageMain.ts

@@ -7,24 +7,30 @@ import { License } from '@/License';
 import { Logger } from '@/Logger';
 import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
 import { Push } from '@/push';
+import { TestWebhooks } from '@/TestWebhooks';
 import { OrchestrationService } from '@/services/orchestration.service';
 import { WorkflowRepository } from '@/databases/repositories/workflow.repository';
-import { TestWebhooks } from '@/TestWebhooks';
 
 export async function handleCommandMessageMain(messageString: string) {
 	const queueModeId = config.getEnv('redis.queueModeId');
 	const isMainInstance = config.getEnv('generic.instanceType') === 'main';
 	const message = messageToRedisServiceCommandObject(messageString);
 	const logger = Container.get(Logger);
-	const activeWorkflowRunner = Container.get(ActiveWorkflowRunner);
 
 	if (message) {
 		logger.debug(
 			`RedisCommandHandler(main): Received command message ${message.command} from ${message.senderId}`,
 		);
+
+		const selfSendingAllowed = [
+			'add-webhooks-triggers-and-pollers',
+			'remove-triggers-and-pollers',
+		].includes(message.command);
+
 		if (
-			message.senderId === queueModeId ||
-			(message.targets && !message.targets.includes(queueModeId))
+			!selfSendingAllowed &&
+			(message.senderId === queueModeId ||
+				(message.targets && !message.targets.includes(queueModeId)))
 		) {
 			// Skipping command message because it's not for this instance
 			logger.debug(
@@ -71,52 +77,106 @@ export async function handleCommandMessageMain(messageString: string) {
 				await Container.get(ExternalSecretsManager).reloadAllProviders();
 				break;
 
-			case 'workflowActiveStateChanged': {
+			case 'add-webhooks-triggers-and-pollers': {
 				if (!debounceMessageReceiver(message, 100)) {
 					message.payload = { result: 'debounced' };
 					return message;
 				}
 
-				const { workflowId, oldState, newState, versionId } = message.payload ?? {};
+				const orchestrationService = Container.get(OrchestrationService);
 
-				if (
-					typeof workflowId !== 'string' ||
-					typeof oldState !== 'boolean' ||
-					typeof newState !== 'boolean' ||
-					typeof versionId !== 'string'
-				) {
-					break;
-				}
+				if (orchestrationService.isFollower) break;
 
-				if (!oldState && newState) {
-					try {
-						await activeWorkflowRunner.add(workflowId, 'activate');
-						push.broadcast('workflowActivated', { workflowId });
-					} catch (e) {
-						const error = e instanceof Error ? e : new Error(`${e}`);
+				if (typeof message.payload?.workflowId !== 'string') break;
 
-						await Container.get(WorkflowRepository).update(workflowId, {
-							active: false,
-							versionId,
+				const { workflowId } = message.payload;
+
+				try {
+					await Container.get(ActiveWorkflowRunner).add(workflowId, 'activate', undefined, {
+						shouldPublish: false, // prevent leader re-publishing message
+					});
+
+					push.broadcast('workflowActivated', { workflowId });
+
+					// instruct followers to show activation in UI
+					await orchestrationService.publish('display-workflow-activation', { workflowId });
+				} catch (error) {
+					if (error instanceof Error) {
+						await Container.get(WorkflowRepository).update(workflowId, { active: false });
+
+						Container.get(Push).broadcast('workflowFailedToActivate', {
+							workflowId,
+							errorMessage: error.message,
 						});
 
-						await Container.get(OrchestrationService).publish('workflowFailedToActivate', {
+						await Container.get(OrchestrationService).publish('workflow-failed-to-activate', {
 							workflowId,
 							errorMessage: error.message,
 						});
 					}
-				} else if (oldState && !newState) {
-					await activeWorkflowRunner.remove(workflowId);
-					push.broadcast('workflowDeactivated', { workflowId });
-				} else {
-					await activeWorkflowRunner.remove(workflowId);
-					await activeWorkflowRunner.add(workflowId, 'update');
 				}
 
-				await activeWorkflowRunner.removeActivationError(workflowId);
+				break;
 			}
 
-			case 'workflowFailedToActivate': {
+			case 'remove-triggers-and-pollers': {
+				if (!debounceMessageReceiver(message, 100)) {
+					message.payload = { result: 'debounced' };
+					return message;
+				}
+
+				const orchestrationService = Container.get(OrchestrationService);
+
+				if (orchestrationService.isFollower) break;
+
+				if (typeof message.payload?.workflowId !== 'string') break;
+
+				const { workflowId } = message.payload;
+
+				const activeWorkflowRunner = Container.get(ActiveWorkflowRunner);
+
+				await activeWorkflowRunner.removeActivationError(workflowId);
+				await activeWorkflowRunner.removeWorkflowTriggersAndPollers(workflowId);
+
+				push.broadcast('workflowDeactivated', { workflowId });
+
+				// instruct followers to show workflow deactivation in UI
+				await orchestrationService.publish('display-workflow-deactivation', { workflowId });
+
+				break;
+			}
+
+			case 'display-workflow-activation': {
+				if (!debounceMessageReceiver(message, 100)) {
+					message.payload = { result: 'debounced' };
+					return message;
+				}
+
+				const { workflowId } = message.payload ?? {};
+
+				if (typeof workflowId !== 'string') break;
+
+				push.broadcast('workflowActivated', { workflowId });
+
+				break;
+			}
+
+			case 'display-workflow-deactivation': {
+				if (!debounceMessageReceiver(message, 100)) {
+					message.payload = { result: 'debounced' };
+					return message;
+				}
+
+				const { workflowId } = message.payload ?? {};
+
+				if (typeof workflowId !== 'string') break;
+
+				push.broadcast('workflowDeactivated', { workflowId });
+
+				break;
+			}
+
+			case 'workflow-failed-to-activate': {
 				if (!debounceMessageReceiver(message, 100)) {
 					message.payload = { result: 'debounced' };
 					return message;

packages/cli/src/services/redis/RedisServiceCommands.ts

@@ -7,8 +7,11 @@ export type RedisServiceCommand =
 	| 'stopWorker'
 	| 'reloadLicense'
 	| 'reloadExternalSecretsProviders'
-	| 'workflowActiveStateChanged' // multi-main only
-	| 'workflowFailedToActivate' // multi-main only
+	| 'display-workflow-activation' // multi-main only
+	| 'display-workflow-deactivation' // multi-main only
+	| 'add-webhooks-triggers-and-pollers' // multi-main only
+	| 'remove-triggers-and-pollers' // multi-main only
+	| 'workflow-failed-to-activate' // multi-main only
 	| 'relay-execution-lifecycle-event' // multi-main only
 	| 'clear-test-webhooks'; // multi-main only