feat(core): Coordinate workflow activation in multiple main scenario in internal API (#7566)

Story: https://linear.app/n8n/issue/PAY-926

This PR coordinates workflow activation on instance startup and on
leadership change in multiple main scenario in the internal API. Part 3
on manual workflow activation and deactivation will be a separate PR.

### Part 1: Instance startup

In multi-main scenario, on starting an instance...
- [x] If the instance is the leader, it should add webhooks, triggers
and pollers.
- [x] If the instance is the follower, it should not add webhooks,
triggers or pollers.
- [x] Unit tests.

### Part 2: Leadership change 

In multi-main scenario, if the main instance leader dies…

- [x] The new main instance leader must activate all trigger- and
poller-based workflows, excluding webhook-based workflows.
- [x] The old main instance leader must deactivate all trigger- and
poller-based workflows, excluding webhook-based workflows.
- [x] Unit tests.

To test, start two instances and check behavior on startup and
leadership change:

```
EXECUTIONS_MODE=queue N8N_LEADER_SELECTION_ENABLED=true N8N_LICENSE_TENANT_ID=... N8N_LICENSE_ACTIVATION_KEY=... N8N_LOG_LEVEL=debug npm run start

EXECUTIONS_MODE=queue N8N_LEADER_SELECTION_ENABLED=true N8N_LICENSE_TENANT_ID=... N8N_LICENSE_ACTIVATION_KEY=... N8N_LOG_LEVEL=debug N8N_PORT=5679 npm run start
```

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 { Service } from 'typedi';
+import Container, { Service } from 'typedi';
 import { ActiveWorkflows, NodeExecuteFunctions } from 'n8n-core';
 
 import type {
@@ -35,8 +35,6 @@ import type express from 'express';
 
 import * as Db from '@/Db';
 import type {
-	IActivationError,
-	IQueuedWorkflowActivations,
 	IResponseCallbackData,
 	IWebhookManager,
 	IWorkflowDb,
@@ -65,99 +63,76 @@ import { webhookNotFoundErrorMessage } from './utils';
 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';
 
 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)";
 
 @Service()
 export class ActiveWorkflowRunner implements IWebhookManager {
-	private activeWorkflows = new ActiveWorkflows();
+	activeWorkflows = new ActiveWorkflows();
 
 	private activationErrors: {
-		[key: string]: IActivationError;
+		[workflowId: string]: {
+			time: number; // ms
+			error: {
+				message: string;
+			};
+		};
 	} = {};
 
-	private queuedWorkflowActivations: {
-		[key: string]: IQueuedWorkflowActivations;
+	private queuedActivations: {
+		[workflowId: string]: {
+			activationMode: WorkflowActivateMode;
+			lastTimeout: number;
+			timeout: NodeJS.Timeout;
+			workflowData: IWorkflowDb;
+		};
 	} = {};
 
+	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,
 	) {}
 
 	async init() {
-		// Get the active workflows from database
+		if (this.isMultiMainScenario) {
+			const { MultiMainInstancePublisher } = await import(
+				'@/services/orchestration/main/MultiMainInstance.publisher.ee'
+			);
 
-		// NOTE
-		// Here I guess we can have a flag on the workflow table like hasTrigger
-		// so instead of pulling all the active webhooks just pull the actives that have a trigger
-		const workflowsData: IWorkflowDb[] = (await Db.collections.Workflow.find({
-			where: { active: true },
-			relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
-		})) as IWorkflowDb[];
+			this.multiMainInstancePublisher = Container.get(MultiMainInstancePublisher);
 
-		if (workflowsData.length !== 0) {
-			this.logger.info(' ================================');
-			this.logger.info('   Start Active Workflows:');
-			this.logger.info(' ================================');
-
-			for (const workflowData of workflowsData) {
-				this.logger.info(`   - ${workflowData.name} (ID: ${workflowData.id})`);
-				this.logger.debug(`Initializing active workflow "${workflowData.name}" (startup)`, {
-					workflowName: workflowData.name,
-					workflowId: workflowData.id,
-				});
-				try {
-					await this.add(workflowData.id, 'init', workflowData);
-					this.logger.verbose(`Successfully started workflow "${workflowData.name}"`, {
-						workflowName: workflowData.name,
-						workflowId: workflowData.id,
-					});
-					this.logger.info('     => Started');
-				} catch (error) {
-					ErrorReporter.error(error);
-					this.logger.info(
-						'     => ERROR: Workflow could not be activated on first try, keep on trying if not an auth issue',
-					);
-
-					this.logger.info(`               ${error.message}`);
-					this.logger.error(
-						`Issue on initial workflow activation try "${workflowData.name}" (startup)`,
-						{
-							workflowName: workflowData.name,
-							workflowId: workflowData.id,
-						},
-					);
-					// eslint-disable-next-line @typescript-eslint/no-unsafe-argument
-					this.executeErrorWorkflow(error, workflowData, 'internal');
-
-					if (!error.message.includes('Authorization')) {
-						// Keep on trying to activate the workflow if not an auth issue
-						this.addQueuedWorkflowActivation('init', workflowData);
-					}
-				}
-			}
-			this.logger.verbose('Finished initializing active workflows (startup)');
+			await this.multiMainInstancePublisher.init();
 		}
 
+		await this.addActiveWorkflows('init');
+
 		await this.externalHooks.run('activeWorkflows.initialized', []);
 		await this.webhookService.populateCache();
 	}
 
 	/**
-	 * Removes all the currently active workflows
+	 * Removes all the currently active workflows from memory.
 	 */
-	async removeAll(): Promise<void> {
+	async removeAll() {
 		let activeWorkflowIds: string[] = [];
 		this.logger.verbose('Call to remove all active workflows received (removeAll)');
 
 		activeWorkflowIds.push(...this.activeWorkflows.allActiveWorkflows());
 
-		const activeWorkflows = await this.getActiveWorkflows();
+		const activeWorkflows = await this.allActiveInStorage();
 		activeWorkflowIds = [...activeWorkflowIds, ...activeWorkflows];
 		// Make sure IDs are unique
 		activeWorkflowIds = Array.from(new Set(activeWorkflowIds));
@@ -284,76 +259,86 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	}
 
 	/**
-	 * Returns the ids of the currently active workflows
+	 * Returns the ids of the currently active workflows from memory.
 	 */
-	async getActiveWorkflows(user?: User): Promise<string[]> {
-		let activeWorkflows: WorkflowEntity[] = [];
-		if (!user || user.globalRole.name === 'owner') {
-			activeWorkflows = await Db.collections.Workflow.find({
-				select: ['id'],
-				where: { active: true },
-			});
-			return activeWorkflows
-				.map((workflow) => workflow.id)
-				.filter((workflowId) => !this.activationErrors[workflowId]);
-		} else {
-			const active = await Db.collections.Workflow.find({
-				select: ['id'],
-				where: { active: true },
-			});
-			const activeIds = active.map((workflow) => workflow.id);
-			const where = whereClause({
-				user,
-				entityType: 'workflow',
-			});
-			Object.assign(where, { workflowId: In(activeIds) });
-			const shared = await Db.collections.SharedWorkflow.find({
-				select: ['workflowId'],
-				where,
-			});
-			return shared
-				.map((id) => id.workflowId)
-				.filter((workflowId) => !this.activationErrors[workflowId]);
-		}
+	allActiveInMemory() {
+		return this.activeWorkflows.allActiveWorkflows();
 	}
 
 	/**
-	 * Returns if the workflow is active
-	 *
-	 * @param {string} id The id of the workflow to check
+	 * Get the IDs of active workflows from storage.
 	 */
-	async isActive(id: string): Promise<boolean> {
-		const workflow = await Db.collections.Workflow.findOne({
-			select: ['active'],
-			where: { id },
+	async allActiveInStorage(user?: User) {
+		const isFullAccess = !user || user.globalRole.name === 'owner';
+
+		if (isFullAccess) {
+			const activeWorkflows = await this.workflowRepository.find({
+				select: ['id'],
+				where: { active: true },
+			});
+
+			return activeWorkflows
+				.map((workflow) => workflow.id)
+				.filter((workflowId) => !this.activationErrors[workflowId]);
+		}
+
+		const where = whereClause({
+			user,
+			entityType: 'workflow',
 		});
+
+		const activeWorkflows = await this.workflowRepository.find({
+			select: ['id'],
+			where: { active: true },
+		});
+
+		const activeIds = activeWorkflows.map((workflow) => workflow.id);
+
+		Object.assign(where, { workflowId: In(activeIds) });
+
+		const sharings = await Db.collections.SharedWorkflow.find({
+			select: ['workflowId'],
+			where,
+		});
+
+		return sharings
+			.map((sharing) => sharing.workflowId)
+			.filter((workflowId) => !this.activationErrors[workflowId]);
+	}
+
+	/**
+	 * Returns if the workflow is stored as `active`.
+	 *
+	 * @important Do not confuse with `ActiveWorkflows.isActive()`,
+	 * which checks if the workflow is active in memory.
+	 */
+	async isActive(workflowId: string) {
+		const workflow = await this.workflowRepository.findOne({
+			select: ['active'],
+			where: { id: workflowId },
+		});
+
 		return !!workflow?.active;
 	}
 
 	/**
 	 * Return error if there was a problem activating the workflow
-	 *
-	 * @param {string} id The id of the workflow to return the error of
 	 */
-	getActivationError(id: string): IActivationError | undefined {
-		if (this.activationErrors[id] === undefined) {
-			return undefined;
-		}
-
-		return this.activationErrors[id];
+	getActivationError(workflowId: string) {
+		return this.activationErrors[workflowId];
 	}
 
 	/**
-	 * Adds all the webhooks of the workflow
+	 * Register workflow-defined webhooks in the `workflow_entity` table.
 	 */
-	async addWorkflowWebhooks(
+	async addWebhooks(
 		workflow: Workflow,
 		additionalData: IWorkflowExecuteAdditionalDataWorkflow,
 		mode: WorkflowExecuteMode,
 		activation: WorkflowActivateMode,
-	): Promise<void> {
+	) {
 		const webhooks = WebhookHelpers.getWorkflowWebhooks(workflow, additionalData, undefined, true);
-		let path = '' as string | undefined;
+		let path = '';
 
 		for (const webhookData of webhooks) {
 			const node = workflow.getNode(webhookData.node) as INode;
@@ -401,7 +386,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 				}
 
 				try {
-					await this.removeWorkflowWebhooks(workflow.id);
+					await this.clearWebhooks(workflow.id);
 				} catch (error1) {
 					ErrorReporter.error(error1);
 					this.logger.error(
@@ -428,14 +413,14 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	}
 
 	/**
-	 * Remove all the webhooks of the workflow
-	 *
+	 * Clear workflow-defined webhooks from the `webhook_entity` table.
 	 */
-	async removeWorkflowWebhooks(workflowId: string): Promise<void> {
+	async clearWebhooks(workflowId: string) {
 		const workflowData = await Db.collections.Workflow.findOne({
 			where: { id: workflowId },
 			relations: ['shared', 'shared.user', 'shared.user.globalRole'],
 		});
+
 		if (workflowData === null) {
 			throw new Error(`Could not find workflow with id "${workflowId}"`);
 		}
@@ -468,11 +453,6 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 		await this.webhookService.deleteWorkflowWebhooks(workflowId);
 	}
 
-	/**
-	 * Runs the given workflow
-	 *
-	 */
-
 	async runWorkflow(
 		workflowData: IWorkflowDb,
 		node: INode,
@@ -520,7 +500,6 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	/**
 	 * Return poll function which gets the global functions from n8n-core
 	 * and overwrites the emit to be able to start it in subprocess
-	 *
 	 */
 	getExecutePollFunctions(
 		workflowData: IWorkflowDb,
@@ -576,7 +555,6 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	/**
 	 * Return trigger function which gets the global functions from n8n-core
 	 * and overwrites the emit to be able to start it in subprocess
-	 *
 	 */
 	getExecuteTriggerFunctions(
 		workflowData: IWorkflowDb,
@@ -647,7 +625,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 				);
 				this.executeErrorWorkflow(activationError, workflowData, mode);
 
-				this.addQueuedWorkflowActivation(activation, workflowData);
+				this.addQueuedWorkflowActivation(activation, workflowData as WorkflowEntity);
 			};
 			return returnFunctions;
 		};
@@ -676,113 +654,175 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	}
 
 	/**
-	 * Makes a workflow active
+	 * Register as active in memory all workflows stored as `active`.
+	 */
+	async addActiveWorkflows(activationMode: WorkflowActivateMode) {
+		const dbWorkflows = await this.workflowRepository.getAllActive();
+
+		if (dbWorkflows.length === 0) return;
+
+		this.logger.info(' ================================');
+		this.logger.info('   Start Active Workflows:');
+		this.logger.info(' ================================');
+
+		for (const dbWorkflow of dbWorkflows) {
+			this.logger.info(`   - ${dbWorkflow.display()}`);
+			this.logger.debug(`Initializing active workflow ${dbWorkflow.display()} (startup)`, {
+				workflowName: dbWorkflow.name,
+				workflowId: dbWorkflow.id,
+			});
+
+			try {
+				await this.add(dbWorkflow.id, activationMode, dbWorkflow);
+
+				this.logger.verbose(`Successfully started workflow ${dbWorkflow.display()}`, {
+					workflowName: dbWorkflow.name,
+					workflowId: dbWorkflow.id,
+				});
+				this.logger.info('     => Started');
+			} catch (error) {
+				ErrorReporter.error(error);
+				this.logger.info(
+					'     => ERROR: Workflow could not be activated on first try, keep on trying if not an auth issue',
+				);
+
+				this.logger.info(`               ${error.message}`);
+				this.logger.error(
+					`Issue on initial workflow activation try of ${dbWorkflow.display()} (startup)`,
+					{
+						workflowName: dbWorkflow.name,
+						workflowId: dbWorkflow.id,
+					},
+				);
+
+				// eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+				this.executeErrorWorkflow(error, dbWorkflow, 'internal');
+
+				// do not keep trying to activate on authorization error
+				if (error.message.includes('Authorization')) continue;
+
+				this.addQueuedWorkflowActivation('init', dbWorkflow);
+			}
+		}
+
+		this.logger.verbose('Finished activating workflows (startup)');
+	}
+
+	async addAllTriggerAndPollerBasedWorkflows() {
+		this.logger.debug('[Leadership change] Adding all trigger- and poller-based workflows...');
+
+		await this.addActiveWorkflows('leadershipChange');
+	}
+
+	async removeAllTriggerAndPollerBasedWorkflows() {
+		this.logger.debug('[Leadership change] Removing all trigger- and poller-based workflows...');
+
+		await this.activeWorkflows.removeAllTriggerAndPollerBasedWorkflows();
+	}
+
+	/**
+	 * Register a workflow as active.
 	 *
-	 * @param {string} workflowId The id of the workflow to activate
-	 * @param {IWorkflowDb} [workflowData] If workflowData is given it saves the DB query
+	 * An activatable workflow may be webhook-, trigger-, or poller-based:
+	 *
+	 * - A `webhook` is an HTTP-based node that can start a workflow when called
+	 * by a third-party service.
+	 * - A `poller` is an HTTP-based node that can start a workflow when detecting
+	 * a change while regularly checking a third-party service.
+	 * - A `trigger` is any non-HTTP-based node that can start a workflow, e.g. a
+	 * time-based node like Schedule Trigger or a message-queue-based node.
+	 *
+	 * Note that despite the name, most "trigger" nodes are actually webhook-based
+	 * and so qualify as `webhook`, e.g. Stripe Trigger.
+	 *
+	 * Triggers and pollers are registered as active in memory at `ActiveWorkflows`,
+	 * but webhooks are registered by being entered in the `webhook_entity` table,
+	 * since webhooks do not require continuous execution.
 	 */
 	async add(
 		workflowId: string,
-		activation: WorkflowActivateMode,
-		workflowData?: IWorkflowDb,
-	): Promise<void> {
-		let workflowInstance: Workflow;
+		activationMode: WorkflowActivateMode,
+		existingWorkflow?: WorkflowEntity,
+	) {
+		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.isMultiMainScenario && activationMode === 'leadershipChange') {
+			shouldAddWebhooks = false;
+			shouldAddTriggersAndPollers = true;
+		}
+
 		try {
-			if (workflowData === undefined) {
-				workflowData = (await Db.collections.Workflow.findOne({
-					where: { id: workflowId },
-					relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
-				})) as IWorkflowDb;
+			const dbWorkflow = existingWorkflow ?? (await this.workflowRepository.findById(workflowId));
+
+			if (!dbWorkflow) {
+				throw new WorkflowActivationError(`Failed to find workflow with ID "${workflowId}"`);
 			}
 
-			if (!workflowData) {
-				throw new Error(`Could not find workflow with id "${workflowId}".`);
-			}
-			workflowInstance = new Workflow({
-				id: workflowId,
-				name: workflowData.name,
-				nodes: workflowData.nodes,
-				connections: workflowData.connections,
-				active: workflowData.active,
+			workflow = new Workflow({
+				id: dbWorkflow.id,
+				name: dbWorkflow.name,
+				nodes: dbWorkflow.nodes,
+				connections: dbWorkflow.connections,
+				active: dbWorkflow.active,
 				nodeTypes: this.nodeTypes,
-				staticData: workflowData.staticData,
-				settings: workflowData.settings,
+				staticData: dbWorkflow.staticData,
+				settings: dbWorkflow.settings,
 			});
 
-			const canBeActivated = workflowInstance.checkIfWorkflowCanBeActivated(STARTING_NODES);
+			const canBeActivated = workflow.checkIfWorkflowCanBeActivated(STARTING_NODES);
+
 			if (!canBeActivated) {
-				this.logger.error(`Unable to activate workflow "${workflowData.name}"`);
-				throw new Error(
-					'The workflow can not be activated because it does not contain any nodes which could start the workflow. Only workflows which have trigger or webhook nodes can be activated.',
+				throw new WorkflowActivationError(
+					`Workflow ${dbWorkflow.display()} has no node to start the workflow - at least one trigger, poller or webhook node is required`,
 				);
 			}
 
-			const mode = 'trigger';
-			const workflowOwner = (workflowData as WorkflowEntity).shared.find(
-				(shared) => shared.role.name === 'owner',
-			);
-			if (!workflowOwner) {
-				throw new Error('Workflow cannot be activated because it has no owner');
+			const sharing = dbWorkflow.shared.find((shared) => shared.role.name === 'owner');
+
+			if (!sharing) {
+				throw new WorkflowActivationError(`Workflow ${dbWorkflow.display()} has no owner`);
 			}
-			const additionalData = await WorkflowExecuteAdditionalData.getBase(workflowOwner.user.id);
-			const getTriggerFunctions = this.getExecuteTriggerFunctions(
-				workflowData,
-				additionalData,
-				mode,
-				activation,
-			);
-			const getPollFunctions = this.getExecutePollFunctions(
-				workflowData,
-				additionalData,
-				mode,
-				activation,
-			);
 
-			// Add the workflows which have webhooks defined
-			await this.addWorkflowWebhooks(workflowInstance, additionalData, mode, activation);
+			const additionalData = await WorkflowExecuteAdditionalData.getBase(sharing.user.id);
 
-			if (
-				workflowInstance.getTriggerNodes().length !== 0 ||
-				workflowInstance.getPollNodes().length !== 0
-			) {
-				await this.activeWorkflows.add(
-					workflowId,
-					workflowInstance,
+			if (shouldAddWebhooks) {
+				this.logger.debug('============');
+				this.logger.debug(`Adding webhooks for workflow "${dbWorkflow.display()}"`);
+				this.logger.debug('============');
+
+				await this.addWebhooks(workflow, additionalData, 'trigger', activationMode);
+			}
+
+			if (shouldAddTriggersAndPollers) {
+				this.logger.debug('============');
+				this.logger.debug(`Adding triggers and pollers for workflow "${dbWorkflow.display()}"`);
+				this.logger.debug('============');
+
+				await this.addTriggersAndPollers(dbWorkflow, workflow, {
+					activationMode,
+					executionMode: 'trigger',
 					additionalData,
-					mode,
-					activation,
-					getTriggerFunctions,
-					getPollFunctions,
-				);
-				this.logger.verbose(`Successfully activated workflow "${workflowData.name}"`, {
-					workflowId,
-					workflowName: workflowData.name,
 				});
 			}
 
 			// Workflow got now successfully activated so make sure nothing is left in the queue
 			this.removeQueuedWorkflowActivation(workflowId);
 
-			if (this.activationErrors[workflowId] !== undefined) {
-				// If there were activation errors delete them
+			if (this.activationErrors[workflowId]) {
 				delete this.activationErrors[workflowId];
 			}
 
-			if (workflowInstance.id) {
-				// Sum all triggers in the workflow, EXCLUDING the manual trigger
-				const triggerFilter = (nodeType: INodeType) =>
-					!!nodeType.trigger && !nodeType.description.name.includes('manualTrigger');
-				const triggerCount =
-					workflowInstance.queryNodes(triggerFilter).length +
-					workflowInstance.getPollNodes().length +
-					WebhookHelpers.getWorkflowWebhooks(workflowInstance, additionalData, undefined, true)
-						.length;
-				await WorkflowsService.updateWorkflowTriggerCount(workflowInstance.id, triggerCount);
-			}
+			const triggerCount = this.countTriggers(workflow, additionalData);
+			await WorkflowsService.updateWorkflowTriggerCount(workflow.id, triggerCount);
 		} catch (error) {
-			// There was a problem activating the workflow
-
-			// Save the error
 			this.activationErrors[workflowId] = {
 				time: new Date().getTime(),
 				error: {
@@ -795,7 +835,24 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 
 		// If for example webhooks get created it sometimes has to save the
 		// id of them in the static data. So make sure that data gets persisted.
-		await WorkflowsService.saveStaticData(workflowInstance!);
+		await WorkflowsService.saveStaticData(workflow);
+	}
+
+	/**
+	 * Count all triggers in the workflow, excluding Manual Trigger.
+	 */
+	private countTriggers(
+		workflow: Workflow,
+		additionalData: IWorkflowExecuteAdditionalDataWorkflow,
+	) {
+		const triggerFilter = (nodeType: INodeType) =>
+			!!nodeType.trigger && !nodeType.description.name.includes('manualTrigger');
+
+		return (
+			workflow.queryNodes(triggerFilter).length +
+			workflow.getPollNodes().length +
+			WebhookHelpers.getWorkflowWebhooks(workflow, additionalData, undefined, true).length
+		);
 	}
 
 	/**
@@ -803,10 +860,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	 * Meaning it will keep on trying to activate it in regular
 	 * amounts indefinitely.
 	 */
-	addQueuedWorkflowActivation(
-		activationMode: WorkflowActivateMode,
-		workflowData: IWorkflowDb,
-	): void {
+	addQueuedWorkflowActivation(activationMode: WorkflowActivateMode, workflowData: WorkflowEntity) {
 		const workflowId = workflowData.id;
 		const workflowName = workflowData.name;
 
@@ -819,7 +873,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 				await this.add(workflowId, activationMode, workflowData);
 			} catch (error) {
 				ErrorReporter.error(error);
-				let lastTimeout = this.queuedWorkflowActivations[workflowId].lastTimeout;
+				let lastTimeout = this.queuedActivations[workflowId].lastTimeout;
 				if (lastTimeout < WORKFLOW_REACTIVATE_MAX_TIMEOUT) {
 					lastTimeout = Math.min(lastTimeout * 2, WORKFLOW_REACTIVATE_MAX_TIMEOUT);
 				}
@@ -834,8 +888,8 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 					},
 				);
 
-				this.queuedWorkflowActivations[workflowId].lastTimeout = lastTimeout;
-				this.queuedWorkflowActivations[workflowId].timeout = setTimeout(retryFunction, lastTimeout);
+				this.queuedActivations[workflowId].lastTimeout = lastTimeout;
+				this.queuedActivations[workflowId].timeout = setTimeout(retryFunction, lastTimeout);
 				return;
 			}
 			this.logger.info(
@@ -851,7 +905,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 		// multiple run in parallel
 		this.removeQueuedWorkflowActivation(workflowId);
 
-		this.queuedWorkflowActivations[workflowId] = {
+		this.queuedActivations[workflowId] = {
 			activationMode,
 			lastTimeout: WORKFLOW_REACTIVATE_INITIAL_TIMEOUT,
 			timeout: setTimeout(retryFunction, WORKFLOW_REACTIVATE_INITIAL_TIMEOUT),
@@ -862,18 +916,18 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	/**
 	 * Remove a workflow from the activation queue
 	 */
-	removeQueuedWorkflowActivation(workflowId: string): void {
-		if (this.queuedWorkflowActivations[workflowId]) {
-			clearTimeout(this.queuedWorkflowActivations[workflowId].timeout);
-			delete this.queuedWorkflowActivations[workflowId];
+	removeQueuedWorkflowActivation(workflowId: string) {
+		if (this.queuedActivations[workflowId]) {
+			clearTimeout(this.queuedActivations[workflowId].timeout);
+			delete this.queuedActivations[workflowId];
 		}
 	}
 
 	/**
 	 * Remove all workflows from the activation queue
 	 */
-	removeAllQueuedWorkflowActivations(): void {
-		for (const workflowId in this.queuedWorkflowActivations) {
+	removeAllQueuedWorkflowActivations() {
+		for (const workflowId in this.queuedActivations) {
 			this.removeQueuedWorkflowActivation(workflowId);
 		}
 	}
@@ -884,10 +938,10 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 	 * @param {string} workflowId The id of the workflow to deactivate
 	 */
 	// TODO: this should happen in a transaction
-	async remove(workflowId: string): Promise<void> {
+	async remove(workflowId: string) {
 		// Remove all the webhooks of the workflow
 		try {
-			await this.removeWorkflowWebhooks(workflowId);
+			await this.clearWebhooks(workflowId);
 		} catch (error) {
 			ErrorReporter.error(error);
 			this.logger.error(
@@ -900,7 +954,7 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 			delete this.activationErrors[workflowId];
 		}
 
-		if (this.queuedWorkflowActivations[workflowId] !== undefined) {
+		if (this.queuedActivations[workflowId] !== undefined) {
 			this.removeQueuedWorkflowActivation(workflowId);
 		}
 
@@ -913,4 +967,52 @@ export class ActiveWorkflowRunner implements IWebhookManager {
 			}
 		}
 	}
+
+	/**
+	 * Register as active in memory a trigger- or poller-based workflow.
+	 */
+	async addTriggersAndPollers(
+		dbWorkflow: WorkflowEntity,
+		workflow: Workflow,
+		{
+			activationMode,
+			executionMode,
+			additionalData,
+		}: {
+			activationMode: WorkflowActivateMode;
+			executionMode: WorkflowExecuteMode;
+			additionalData: IWorkflowExecuteAdditionalDataWorkflow;
+		},
+	) {
+		const getTriggerFunctions = this.getExecuteTriggerFunctions(
+			dbWorkflow,
+			additionalData,
+			executionMode,
+			activationMode,
+		);
+
+		const getPollFunctions = this.getExecutePollFunctions(
+			dbWorkflow,
+			additionalData,
+			executionMode,
+			activationMode,
+		);
+
+		if (workflow.getTriggerNodes().length !== 0 || workflow.getPollNodes().length !== 0) {
+			await this.activeWorkflows.add(
+				workflow.id,
+				workflow,
+				additionalData,
+				executionMode,
+				activationMode,
+				getTriggerFunctions,
+				getPollFunctions,
+			);
+
+			this.logger.verbose(`Workflow ${dbWorkflow.display()} activated`, {
+				workflowId: dbWorkflow.id,
+				workflowName: dbWorkflow.name,
+			});
+		}
+	}
 }

packages/cli/src/Interfaces.ts

@@ -16,7 +16,6 @@ import type {
 	IWorkflowBase,
 	CredentialLoadingDetails,
 	Workflow,
-	WorkflowActivateMode,
 	WorkflowExecuteMode,
 	ExecutionStatus,
 	IExecutionsSummary,
@@ -64,20 +63,6 @@ import type {
 import type { LICENSE_FEATURES, LICENSE_QUOTAS } from './constants';
 import type { WorkflowWithSharingsAndCredentials } from './workflows/workflows.types';
 
-export interface IActivationError {
-	time: number;
-	error: {
-		message: string;
-	};
-}
-
-export interface IQueuedWorkflowActivations {
-	activationMode: WorkflowActivateMode;
-	lastTimeout: number;
-	timeout: NodeJS.Timeout;
-	workflowData: IWorkflowDb;
-}
-
 export interface ICredentialsTypeData {
 	[key: string]: CredentialLoadingDetails;
 }

packages/cli/src/Server.ts

@@ -620,7 +620,7 @@ export class Server extends AbstractServer {
 		this.app.get(
 			`/${this.restEndpoint}/active`,
 			ResponseHelper.send(async (req: WorkflowRequest.GetAllActive) => {
-				return this.activeWorkflowRunner.getActiveWorkflows(req.user);
+				return this.activeWorkflowRunner.allActiveInStorage(req.user);
 			}),
 		);
 

packages/cli/src/commands/start.ts

@@ -119,6 +119,8 @@ export class Start extends BaseCommand {
 					'@/services/orchestration/main/MultiMainInstance.publisher.ee'
 				);
 
+				await this.activeWorkflowRunner.removeAllTriggerAndPollerBasedWorkflows();
+
 				await Container.get(MultiMainInstancePublisher).destroy();
 			}
 
@@ -251,6 +253,15 @@ export class Start extends BaseCommand {
 		}
 
 		await Container.get(OrchestrationHandlerMainService).init();
+
+		multiMainInstancePublisher.on('leadershipChange', async () => {
+			if (multiMainInstancePublisher.isLeader) {
+				await this.activeWorkflowRunner.addAllTriggerAndPollerBasedWorkflows();
+			} else {
+				// only in case of leadership change without shutdown
+				await this.activeWorkflowRunner.removeAllTriggerAndPollerBasedWorkflows();
+			}
+		});
 	}
 
 	async run() {

packages/cli/src/databases/entities/WorkflowEntity.ts

@@ -89,6 +89,10 @@ export class WorkflowEntity extends WithTimestampsAndStringId implements IWorkfl
 
 	@Column({ default: 0 })
 	triggerCount: number;
+
+	display() {
+		return `"${this.name}" (ID: ${this.id})`;
+	}
 }
 
 /**

packages/cli/src/databases/repositories/workflow.repository.ts

@@ -7,4 +7,18 @@ export class WorkflowRepository extends Repository<WorkflowEntity> {
 	constructor(dataSource: DataSource) {
 		super(WorkflowEntity, dataSource.manager);
 	}
+
+	async getAllActive() {
+		return this.find({
+			where: { active: true },
+			relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
+		});
+	}
+
+	async findById(workflowId: string) {
+		return this.findOne({
+			where: { id: workflowId },
+			relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
+		});
+	}
 }

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

@@ -2,8 +2,9 @@ import Container from 'typedi';
 import { RedisService } from './redis.service';
 import type { RedisServicePubSubPublisher } from './redis/RedisServicePubSubPublisher';
 import config from '@/config';
+import { EventEmitter } from 'node:events';
 
-export abstract class OrchestrationService {
+export abstract class OrchestrationService extends EventEmitter {
 	protected initialized = false;
 
 	protected queueModeId: string;
@@ -29,6 +30,7 @@ export abstract class OrchestrationService {
 	}
 
 	constructor() {
+		super();
 		this.redisService = Container.get(RedisService);
 		this.queueModeId = config.getEnv('redis.queueModeId');
 	}

packages/cli/src/services/orchestration/main/MultiMainInstance.publisher.ee.ts

@@ -13,11 +13,11 @@ export class MultiMainInstancePublisher extends SingleMainInstancePublisher {
 
 	private leaderId: string | undefined;
 
-	public get isLeader() {
+	get isLeader() {
 		return this.id === this.leaderId;
 	}
 
-	public get isFollower() {
+	get isFollower() {
 		return !this.isLeader;
 	}
 
@@ -84,6 +84,8 @@ export class MultiMainInstancePublisher extends SingleMainInstancePublisher {
 
 			this.leaderId = this.id;
 
+			this.emit('leadershipChange', this.id);
+
 			await this.redisPublisher.setExpiration(this.leaderKey, this.leaderKeyTtl);
 		}
 	}