feat(core): Set up leader selection for multiple main instances (#7527)

https://linear.app/n8n/issue/PAY-933/set-up-leader-selection-for-multiple-main-instances

- [x] Set up new envs
- [x] Add config and license checks
- [x] Implement `MultiMainInstancePublisher`
- [x] Expand `RedisServicePubSubPublisher` to support
`MultiMainInstancePublisher`
- [x] Init `MultiMainInstancePublisher` on startup and destroy on
shutdown
- [x] Add to sandbox plans
- [x] Test manually

Note: This is only for setup - coordinating in reaction to leadership
changes will come in later PRs.

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

@@ -6,6 +6,8 @@ import config from '@/config';
 export abstract class OrchestrationService {
 	protected initialized = false;
 
+	protected queueModeId: string;
+
 	redisPublisher: RedisServicePubSubPublisher;
 
 	readonly redisService: RedisService;
@@ -28,6 +30,7 @@ export abstract class OrchestrationService {
 
 	constructor() {
 		this.redisService = Container.get(RedisService);
+		this.queueModeId = config.getEnv('redis.queueModeId');
 	}
 
 	sanityCheck(): boolean {
@@ -44,7 +47,7 @@ export abstract class OrchestrationService {
 		this.initialized = false;
 	}
 
-	private async initPublisher() {
+	protected async initPublisher() {
 		this.redisPublisher = await this.redisService.getPubSubPublisher();
 	}
 }

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

@@ -0,0 +1,84 @@
+import config from '@/config';
+import { Service } from 'typedi';
+import { TIME } from '@/constants';
+import { SingleMainInstancePublisher } from '@/services/orchestration/main/SingleMainInstance.publisher';
+import { getRedisPrefix } from '@/services/redis/RedisServiceHelper';
+
+/**
+ * For use in main instance, in multiple main instances cluster.
+ */
+@Service()
+export class MultiMainInstancePublisher extends SingleMainInstancePublisher {
+	private id = this.queueModeId;
+
+	private leaderId: string | undefined;
+
+	private get isLeader() {
+		return this.id === this.leaderId;
+	}
+
+	private readonly leaderKey = getRedisPrefix() + ':main_instance_leader';
+
+	private readonly leaderKeyTtl = config.getEnv('leaderSelection.ttl');
+
+	private leaderCheckInterval: NodeJS.Timer | undefined;
+
+	async init() {
+		await this.initPublisher();
+
+		this.initialized = true;
+
+		await this.tryBecomeLeader();
+
+		this.leaderCheckInterval = setInterval(
+			async () => {
+				await this.checkLeader();
+			},
+			config.getEnv('leaderSelection.interval') * TIME.SECOND,
+		);
+	}
+
+	async destroy() {
+		clearInterval(this.leaderCheckInterval);
+
+		if (this.isLeader) await this.redisPublisher.clear(this.leaderKey);
+	}
+
+	private async checkLeader() {
+		if (!this.redisPublisher.redisClient) return;
+
+		const leaderId = await this.redisPublisher.get(this.leaderKey);
+
+		if (!leaderId) {
+			this.logger.debug('Leadership vacant, attempting to become leader...');
+			await this.tryBecomeLeader();
+
+			return;
+		}
+
+		if (this.isLeader) {
+			this.logger.debug(`Leader is this instance "${this.id}"`);
+
+			await this.redisPublisher.setExpiration(this.leaderKey, this.leaderKeyTtl);
+		} else {
+			this.logger.debug(`Leader is other instance "${leaderId}"`);
+
+			this.leaderId = leaderId;
+		}
+	}
+
+	private async tryBecomeLeader() {
+		if (this.isLeader || !this.redisPublisher.redisClient) return;
+
+		// this can only succeed if leadership is currently vacant
+		const keySetSuccessfully = await this.redisPublisher.setIfNotExists(this.leaderKey, this.id);
+
+		if (keySetSuccessfully) {
+			this.logger.debug(`Leader is now this instance "${this.id}"`);
+
+			this.leaderId = this.id;
+
+			await this.redisPublisher.setExpiration(this.leaderKey, this.leaderKeyTtl);
+		}
+	}
+}

packages/cli/src/services/orchestration/main/SingleMainInstance.publisher.ts

@@ -0,0 +1,60 @@
+import { Logger } from '@/Logger';
+import { Service } from 'typedi';
+import { OrchestrationService } from '@/services/orchestration.base.service';
+
+/**
+ * For use in main instance, in single main instance scenario.
+ */
+@Service()
+export class SingleMainInstancePublisher extends OrchestrationService {
+	constructor(protected readonly logger: Logger) {
+		super();
+	}
+
+	sanityCheck() {
+		return this.initialized && this.isQueueMode && this.isMainInstance;
+	}
+
+	async getWorkerStatus(id?: string) {
+		if (!this.sanityCheck()) return;
+
+		const command = 'getStatus';
+
+		this.logger.debug(`Sending "${command}" to command channel`);
+
+		await this.redisPublisher.publishToCommandChannel({
+			command,
+			targets: id ? [id] : undefined,
+		});
+	}
+
+	async getWorkerIds() {
+		if (!this.sanityCheck()) return;
+
+		const command = 'getId';
+
+		this.logger.debug(`Sending "${command}" to command channel`);
+
+		await this.redisPublisher.publishToCommandChannel({ command });
+	}
+
+	async broadcastRestartEventbusAfterDestinationUpdate() {
+		if (!this.sanityCheck()) return;
+
+		const command = 'restartEventBus';
+
+		this.logger.debug(`Sending "${command}" to command channel`);
+
+		await this.redisPublisher.publishToCommandChannel({ command });
+	}
+
+	async broadcastReloadExternalSecretsProviders() {
+		if (!this.sanityCheck()) return;
+
+		const command = 'reloadExternalSecretsProviders';
+
+		this.logger.debug(`Sending "${command}" to command channel`);
+
+		await this.redisPublisher.publishToCommandChannel({ command });
+	}
+}

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

@@ -34,7 +34,8 @@ export async function handleCommandMessageMain(messageString: string) {
 					};
 					return message;
 				}
-				if (isMainInstance) {
+
+				if (isMainInstance && !config.getEnv('leaderSelection.enabled')) {
 					// at this point in time, only a single main instance is supported, thus this command _should_ never be caught currently
 					logger.error(
 						'Received command to reload license via Redis, but this should not have happened and is not supported on the main instance yet.',

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

@@ -1,38 +0,0 @@
-import { Service } from 'typedi';
-import { OrchestrationService } from '../../orchestration.base.service';
-
-@Service()
-export class OrchestrationMainService extends OrchestrationService {
-	sanityCheck(): boolean {
-		return this.initialized && this.isQueueMode && this.isMainInstance;
-	}
-
-	async getWorkerStatus(id?: string) {
-		if (!this.sanityCheck()) return;
-		await this.redisPublisher.publishToCommandChannel({
-			command: 'getStatus',
-			targets: id ? [id] : undefined,
-		});
-	}
-
-	async getWorkerIds() {
-		if (!this.sanityCheck()) return;
-		await this.redisPublisher.publishToCommandChannel({
-			command: 'getId',
-		});
-	}
-
-	async broadcastRestartEventbusAfterDestinationUpdate() {
-		if (!this.sanityCheck()) return;
-		await this.redisPublisher.publishToCommandChannel({
-			command: 'restartEventBus',
-		});
-	}
-
-	async broadcastReloadExternalSecretsProviders() {
-		if (!this.sanityCheck()) return;
-		await this.redisPublisher.publishToCommandChannel({
-			command: 'reloadExternalSecretsProviders',
-		});
-	}
-}

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

@@ -39,4 +39,30 @@ export class RedisServicePubSubPublisher extends RedisServiceBaseSender {
 	async publishToWorkerChannel(message: RedisServiceWorkerResponseObject): Promise<void> {
 		await this.publish(WORKER_RESPONSE_REDIS_CHANNEL, JSON.stringify(message));
 	}
+
+	async setIfNotExists(key: string, value: string) {
+		if (!this.redisClient) await this.init();
+
+		const success = await this.redisClient?.setnx(key, value);
+
+		return !!success;
+	}
+
+	async setExpiration(key: string, ttl: number) {
+		if (!this.redisClient) await this.init();
+
+		await this.redisClient?.expire(key, ttl);
+	}
+
+	async get(key: string) {
+		if (!this.redisClient) await this.init();
+
+		return this.redisClient?.get(key);
+	}
+
+	async clear(key: string) {
+		if (!this.redisClient) await this.init();
+
+		await this.redisClient?.del(key);
+	}
 }