refactor(core): Decouple leadership-change handlers using decorators (#15009)

packages/cli/src/__tests__/wait-tracker.test.ts

@@ -7,7 +7,6 @@ import type { ActiveExecutions } from '@/active-executions';
 import type { Project } from '@/databases/entities/project';
 import type { ExecutionRepository } from '@/databases/repositories/execution.repository';
 import type { MultiMainSetup } from '@/scaling/multi-main-setup.ee';
-import { OrchestrationService } from '@/services/orchestration.service';
 import type { OwnershipService } from '@/services/ownership.service';
 import type { IExecutionResponse } from '@/types-db';
 import { WaitTracker } from '@/wait-tracker';
@@ -22,7 +21,6 @@ describe('WaitTracker', () => {
 	const workflowRunner = mock<WorkflowRunner>();
 	const executionRepository = mock<ExecutionRepository>();
 	const multiMainSetup = mock<MultiMainSetup>();
-	const orchestrationService = new OrchestrationService(mock(), multiMainSetup, mock());
 	const instanceSettings = mock<InstanceSettings>({ isLeader: true, isMultiMain: false });
 
 	const project = mock<Project>({ id: 'projectId' });
@@ -46,7 +44,6 @@ describe('WaitTracker', () => {
 			ownershipService,
 			activeExecutions,
 			workflowRunner,
-			orchestrationService,
 			instanceSettings,
 		);
 		multiMainSetup.on.mockReturnThis();
@@ -235,7 +232,6 @@ describe('WaitTracker', () => {
 				ownershipService,
 				activeExecutions,
 				workflowRunner,
-				orchestrationService,
 				mock<InstanceSettings>({ isLeader: false, isMultiMain: false }),
 			);
 

packages/cli/src/active-workflow-manager.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 import { WorkflowsConfig } from '@n8n/config';
-import { OnShutdown } from '@n8n/decorators';
+import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import { chunk } from 'lodash';
 import {
@@ -433,7 +433,7 @@ export class ActiveWorkflowManager {
 			await Promise.all(activationPromises);
 		}
 
-		this.logger.debug('Finished activating workflows (startup)');
+		this.logger.debug('Activated all trigger- and poller-based workflows');
 	}
 
 	private async activateWorkflow(
@@ -487,16 +487,14 @@ export class ActiveWorkflowManager {
 		await this.activationErrorsService.clearAll();
 	}
 
+	@OnLeaderTakeover()
 	async addAllTriggerAndPollerBasedWorkflows() {
-		this.logger.debug('Adding all trigger- and poller-based workflows');
-
 		await this.addActiveWorkflows('leadershipChange');
 	}
 
+	@OnLeaderStepdown()
 	@OnShutdown()
 	async removeAllTriggerAndPollerBasedWorkflows() {
-		this.logger.debug('Removing all trigger- and poller-based workflows');
-
 		await this.activeWorkflows.removeAllTriggerAndPollerBasedWorkflows();
 	}
 

packages/cli/src/commands/start.ts

@@ -261,18 +261,6 @@ export class Start extends BaseCommand {
 		await subscriber.subscribe('n8n.worker-response');
 
 		this.logger.scoped(['scaling', 'pubsub']).debug('Pubsub setup completed');
-
-		if (this.instanceSettings.isSingleMain) return;
-
-		orchestrationService.multiMainSetup
-			.on('leader-stepdown', async () => {
-				this.license.disableAutoRenewals();
-				await this.activeWorkflowManager.removeAllTriggerAndPollerBasedWorkflows();
-			})
-			.on('leader-takeover', async () => {
-				this.license.enableAutoRenewals();
-				await this.activeWorkflowManager.addAllTriggerAndPollerBasedWorkflows();
-			});
 	}
 
 	async run() {

packages/cli/src/license.ts

@@ -6,7 +6,7 @@ import {
 	type BooleanLicenseFeature,
 	type NumericLicenseFeature,
 } from '@n8n/constants';
-import { OnShutdown } from '@n8n/decorators';
+import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Container, Service } from '@n8n/di';
 import type { TEntitlement, TFeatures, TLicenseBlock } from '@n8n_io/license-sdk';
 import { LicenseManager } from '@n8n_io/license-sdk';
@@ -448,10 +448,12 @@ export class License {
 		}
 	}
 
+	@OnLeaderTakeover()
 	enableAutoRenewals() {
 		this.manager?.enableAutoRenewals();
 	}
 
+	@OnLeaderStepdown()
 	disableAutoRenewals() {
 		this.manager?.disableAutoRenewals();
 	}

packages/cli/src/modules/insights/__tests__/insights-collection.service.test.ts

@@ -18,6 +18,7 @@ import type { TypeUnit } from '@/modules/insights/database/entities/insights-sha
 import { InsightsMetadataRepository } from '@/modules/insights/database/repositories/insights-metadata.repository';
 import { InsightsRawRepository } from '@/modules/insights/database/repositories/insights-raw.repository';
 import type { IWorkflowDb } from '@/types-db';
+import { mockLogger } from '@test/mocking';
 import { createTeamProject } from '@test-integration/db/projects';
 import { createWorkflow } from '@test-integration/db/workflows';
 import * as testDb from '@test-integration/test-db';
@@ -284,7 +285,7 @@ describe('workflowExecuteAfterHandler - cacheMetadata', () => {
 		insightsCollectionService = new InsightsCollectionService(
 			sharedWorkflowRepositoryMock,
 			Container.get(InsightsConfig),
-			mock<Logger>(),
+			mockLogger(),
 		);
 	});
 

packages/cli/src/modules/insights/__tests__/insights.service.test.ts

@@ -2,7 +2,6 @@ import type { InsightsDateRange } from '@n8n/api-types';
 import { Container } from '@n8n/di';
 import { mock } from 'jest-mock-extended';
 import { DateTime } from 'luxon';
-import type { Logger } from 'n8n-core';
 
 import type { Project } from '@/databases/entities/project';
 import type { WorkflowEntity } from '@/databases/entities/workflow-entity';
@@ -502,7 +501,6 @@ describe('getAvailableDateRanges', () => {
 			mock<InsightsCompactionService>(),
 			mock<InsightsCollectionService>(),
 			licenseMock,
-			mock<Logger>(),
 		);
 	});
 
@@ -603,7 +601,6 @@ describe('getMaxAgeInDaysAndGranularity', () => {
 			mock<InsightsCompactionService>(),
 			mock<InsightsCollectionService>(),
 			licenseMock,
-			mock<Logger>(),
 		);
 	});
 

packages/cli/src/modules/insights/insights-collection.service.ts

@@ -77,12 +77,14 @@ export class InsightsCollectionService {
 			async () => await this.flushEvents(),
 			this.insightsConfig.flushIntervalSeconds * 1000,
 		);
+		this.logger.debug('Started flushing timer');
 	}
 
 	stopFlushingTimer() {
 		if (this.flushInsightsRawBufferTimer !== undefined) {
 			clearTimeout(this.flushInsightsRawBufferTimer);
 			this.flushInsightsRawBufferTimer = undefined;
+			this.logger.debug('Stopped flushing timer');
 		}
 	}
 

packages/cli/src/modules/insights/insights-compaction.service.ts

@@ -1,4 +1,5 @@
 import { Service } from '@n8n/di';
+import { Logger } from 'n8n-core';
 
 import { InsightsByPeriodRepository } from './database/repositories/insights-by-period.repository';
 import { InsightsRawRepository } from './database/repositories/insights-raw.repository';
@@ -16,7 +17,10 @@ export class InsightsCompactionService {
 		private readonly insightsByPeriodRepository: InsightsByPeriodRepository,
 		private readonly insightsRawRepository: InsightsRawRepository,
 		private readonly insightsConfig: InsightsConfig,
-	) {}
+		private readonly logger: Logger,
+	) {
+		this.logger = this.logger.scoped('insights');
+	}
 
 	startCompactionTimer() {
 		this.stopCompactionTimer();
@@ -24,12 +28,14 @@ export class InsightsCompactionService {
 			async () => await this.compactInsights(),
 			this.insightsConfig.compactionIntervalMinutes * 60 * 1000,
 		);
+		this.logger.debug('Started compaction timer');
 	}
 
 	stopCompactionTimer() {
 		if (this.compactInsightsTimer !== undefined) {
 			clearInterval(this.compactInsightsTimer);
 			this.compactInsightsTimer = undefined;
+			this.logger.debug('Stopped compaction timer');
 		}
 	}
 

packages/cli/src/modules/insights/insights.service.ts

@@ -5,7 +5,6 @@ import {
 } from '@n8n/api-types';
 import { OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
-import { Logger } from 'n8n-core';
 import { UserError } from 'n8n-workflow';
 
 import { License } from '@/license';
@@ -33,19 +32,16 @@ export class InsightsService {
 		private readonly compactionService: InsightsCompactionService,
 		private readonly collectionService: InsightsCollectionService,
 		private readonly license: License,
-		private readonly logger: Logger,
 	) {}
 
 	startBackgroundProcess() {
 		this.compactionService.startCompactionTimer();
 		this.collectionService.startFlushingTimer();
-		this.logger.debug('Started compaction and flushing schedulers');
 	}
 
 	stopBackgroundProcess() {
 		this.compactionService.stopCompactionTimer();
 		this.collectionService.stopFlushingTimer();
-		this.logger.debug('Stopped compaction and flushing schedulers');
 	}
 
 	@OnShutdown()

packages/cli/src/scaling/__tests__/scaling.service.test.ts

@@ -81,7 +81,6 @@ describe('ScalingService', () => {
 			mock(),
 			instanceSettings,
 			mock(),
-			mock(),
 		);
 
 		getRunningJobsCountSpy = jest.spyOn(scalingService, 'getRunningJobsCount');

packages/cli/src/scaling/scaling.service.ts

@@ -1,5 +1,5 @@
 import { GlobalConfig } from '@n8n/config';
-import { OnShutdown } from '@n8n/decorators';
+import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Container, Service } from '@n8n/di';
 import { ErrorReporter, InstanceSettings, isObjectLiteral, Logger } from 'n8n-core';
 import {
@@ -18,7 +18,6 @@ import config from '@/config';
 import { HIGHEST_SHUTDOWN_PRIORITY, Time } from '@/constants';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
 import { EventService } from '@/events/event.service';
-import { OrchestrationService } from '@/services/orchestration.service';
 import { assertNever } from '@/utils';
 
 import { JOB_TYPE_NAME, QUEUE_NAME } from './constants';
@@ -47,7 +46,6 @@ export class ScalingService {
 		private readonly globalConfig: GlobalConfig,
 		private readonly executionRepository: ExecutionRepository,
 		private readonly instanceSettings: InstanceSettings,
-		private readonly orchestrationService: OrchestrationService,
 		private readonly eventService: EventService,
 	) {
 		this.logger = this.logger.scoped('scaling');
@@ -71,15 +69,7 @@ export class ScalingService {
 
 		this.registerListeners();
 
-		const { isLeader, isMultiMain } = this.instanceSettings;
-
-		if (isLeader) this.scheduleQueueRecovery();
-
-		if (isMultiMain) {
-			this.orchestrationService.multiMainSetup
-				.on('leader-takeover', () => this.scheduleQueueRecovery())
-				.on('leader-stepdown', () => this.stopQueueRecovery());
-		}
+		if (this.instanceSettings.isLeader) this.scheduleQueueRecovery();
 
 		this.scheduleQueueMetrics();
 
@@ -434,6 +424,7 @@ export class ScalingService {
 		waitMs: config.getEnv('executions.queueRecovery.interval') * 60 * 1000,
 	};
 
+	@OnLeaderTakeover()
 	private scheduleQueueRecovery(waitMs = this.queueRecoveryContext.waitMs) {
 		this.queueRecoveryContext.timeout = setTimeout(async () => {
 			try {
@@ -454,7 +445,10 @@ export class ScalingService {
 		this.logger.debug(`Scheduled queue recovery check for next ${wait}`);
 	}
 
+	@OnLeaderStepdown()
 	private stopQueueRecovery() {
+		if (!this.queueRecoveryContext.timeout) return;
+
 		clearTimeout(this.queueRecoveryContext.timeout);
 
 		this.logger.debug('Queue recovery stopped');

packages/cli/src/services/pruning/__tests__/pruning.service.test.ts

@@ -2,8 +2,6 @@ import type { ExecutionsConfig } from '@n8n/config';
 import { mock } from 'jest-mock-extended';
 import type { InstanceSettings } from 'n8n-core';
 
-import type { MultiMainSetup } from '@/scaling/multi-main-setup.ee';
-import type { OrchestrationService } from '@/services/orchestration.service';
 import { mockLogger } from '@test/mocking';
 
 import { PruningService } from '../pruning.service';
@@ -20,9 +18,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: true, isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock(),
 			);
 			const startPruningSpy = jest.spyOn(pruningService, 'startPruning');
@@ -38,9 +33,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: false, isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock(),
 			);
 			const startPruningSpy = jest.spyOn(pruningService, 'startPruning');
@@ -49,33 +41,6 @@ describe('PruningService', () => {
 
 			expect(startPruningSpy).not.toHaveBeenCalled();
 		});
-
-		it('should register leadership events if main on multi-main setup', () => {
-			const pruningService = new PruningService(
-				mockLogger(),
-				mock<InstanceSettings>({ isLeader: true, isMultiMain: true }),
-				mock(),
-				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>({ on: jest.fn() }),
-				}),
-				mock(),
-			);
-
-			pruningService.init();
-
-			// @ts-expect-error Private method
-			expect(pruningService.orchestrationService.multiMainSetup.on).toHaveBeenCalledWith(
-				'leader-takeover',
-				expect.any(Function),
-			);
-
-			// @ts-expect-error Private method
-			expect(pruningService.orchestrationService.multiMainSetup.on).toHaveBeenCalledWith(
-				'leader-stepdown',
-				expect.any(Function),
-			);
-		});
 	});
 
 	describe('isEnabled', () => {
@@ -85,9 +50,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: true }),
 			);
 
@@ -100,9 +62,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: false }),
 			);
 
@@ -115,9 +74,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: false, instanceType: 'worker', isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: true }),
 			);
 
@@ -135,9 +91,6 @@ describe('PruningService', () => {
 				}),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: true }),
 			);
 
@@ -152,9 +105,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: false }),
 			);
 
@@ -179,9 +129,6 @@ describe('PruningService', () => {
 				mock<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
 				mock(),
 				mock(),
-				mock<OrchestrationService>({
-					multiMainSetup: mock<MultiMainSetup>(),
-				}),
 				mock<ExecutionsConfig>({ pruneData: true }),
 			);
 

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

@@ -1,5 +1,5 @@
 import { ExecutionsConfig } from '@n8n/config';
-import { OnShutdown } from '@n8n/decorators';
+import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import { BinaryDataService, InstanceSettings, Logger } from 'n8n-core';
 import { ensureError } from 'n8n-workflow';
@@ -9,8 +9,6 @@ import { Time } from '@/constants';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
 import { connectionState as dbConnectionState } from '@/db';
 
-import { OrchestrationService } from '../orchestration.service';
-
 /**
  * Responsible for deleting old executions from the database and deleting their
  * associated binary data from the filesystem, on a rolling basis.
@@ -47,7 +45,6 @@ export class PruningService {
 		private readonly instanceSettings: InstanceSettings,
 		private readonly executionRepository: ExecutionRepository,
 		private readonly binaryDataService: BinaryDataService,
-		private readonly orchestrationService: OrchestrationService,
 		private readonly executionsConfig: ExecutionsConfig,
 	) {
 		this.logger = this.logger.scoped('pruning');
@@ -57,11 +54,6 @@ export class PruningService {
 		strict(this.instanceSettings.instanceRole !== 'unset', 'Instance role is not set');
 
 		if (this.instanceSettings.isLeader) this.startPruning();
-
-		if (this.instanceSettings.isMultiMain) {
-			this.orchestrationService.multiMainSetup.on('leader-takeover', () => this.startPruning());
-			this.orchestrationService.multiMainSetup.on('leader-stepdown', () => this.stopPruning());
-		}
 	}
 
 	get isEnabled() {
@@ -72,18 +64,24 @@ export class PruningService {
 		);
 	}
 
+	@OnLeaderTakeover()
 	startPruning() {
 		if (!this.isEnabled || !dbConnectionState.migrated || this.isShuttingDown) return;
 
 		this.scheduleRollingSoftDeletions();
 		this.scheduleNextHardDeletion();
+
+		this.logger.debug('Started pruning timers');
 	}
 
+	@OnLeaderStepdown()
 	stopPruning() {
 		if (!this.isEnabled) return;
 
 		clearInterval(this.softDeletionInterval);
 		clearTimeout(this.hardDeletionTimeout);
+
+		this.logger.debug('Stopped pruning timers');
 	}
 
 	private scheduleRollingSoftDeletions(rateMs = this.rates.softDeletion) {

packages/cli/src/wait-tracker.ts

@@ -1,10 +1,10 @@
+import { OnLeaderStepdown, OnLeaderTakeover } from '@n8n/decorators';
 import { Service } from '@n8n/di';
 import { InstanceSettings, Logger } from 'n8n-core';
 import { UnexpectedError, type IWorkflowExecutionDataProcess } from 'n8n-workflow';
 
 import { ActiveExecutions } from '@/active-executions';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
-import { OrchestrationService } from '@/services/orchestration.service';
 import { OwnershipService } from '@/services/ownership.service';
 import { WorkflowRunner } from '@/workflow-runner';
 
@@ -25,7 +25,6 @@ export class WaitTracker {
 		private readonly ownershipService: OwnershipService,
 		private readonly activeExecutions: ActiveExecutions,
 		private readonly workflowRunner: WorkflowRunner,
-		private readonly orchestrationService: OrchestrationService,
 		private readonly instanceSettings: InstanceSettings,
 	) {
 		this.logger = this.logger.scoped('waiting-executions');
@@ -35,30 +34,20 @@ export class WaitTracker {
 		return this.waitingExecutions[executionId] !== undefined;
 	}
 
-	/**
-	 * @important Requires `OrchestrationService` to be initialized.
-	 */
 	init() {
-		const { isLeader, isMultiMain } = this.instanceSettings;
-
-		if (isLeader) this.startTracking();
-
-		if (isMultiMain) {
-			this.orchestrationService.multiMainSetup
-				.on('leader-takeover', () => this.startTracking())
-				.on('leader-stepdown', () => this.stopTracking());
-		}
+		if (this.instanceSettings.isLeader) this.startTracking();
 	}
 
+	@OnLeaderTakeover()
 	private startTracking() {
-		this.logger.debug('Started tracking waiting executions');
-
 		// Poll every 60 seconds a list of upcoming executions
 		this.mainTimer = setInterval(() => {
 			void this.getWaitingExecutions();
 		}, 60000);
 
 		void this.getWaitingExecutions();
+
+		this.logger.debug('Started tracking waiting executions');
 	}
 
 	async getWaitingExecutions() {
@@ -143,12 +132,15 @@ export class WaitTracker {
 		}
 	}
 
+	@OnLeaderStepdown()
 	stopTracking() {
-		this.logger.debug('Shutting down wait tracking');
+		if (!this.mainTimer) return;
 
 		clearInterval(this.mainTimer);
 		Object.keys(this.waitingExecutions).forEach((executionId) => {
 			clearTimeout(this.waitingExecutions[executionId].timer);
 		});
+
+		this.logger.debug('Stopped tracking waiting executions');
 	}
 }

packages/cli/test/integration/pruning.service.test.ts

@@ -1,6 +1,5 @@
 import { ExecutionsConfig } from '@n8n/config';
 import { Container } from '@n8n/di';
-import { mock } from 'jest-mock-extended';
 import { BinaryDataService, InstanceSettings } from 'n8n-core';
 import type { ExecutionStatus, IWorkflowBase } from 'n8n-workflow';
 
@@ -37,7 +36,6 @@ describe('softDeleteOnPruningCycle()', () => {
 			instanceSettings,
 			Container.get(ExecutionRepository),
 			mockInstance(BinaryDataService),
-			mock(),
 			executionsConfig,
 		);
 

packages/core/src/execution-engine/active-workflows.ts

@@ -211,9 +211,17 @@ export class ActiveWorkflows {
 	}
 
 	async removeAllTriggerAndPollerBasedWorkflows() {
-		for (const workflowId of Object.keys(this.activeWorkflows)) {
+		const activeWorkflowIds = Object.keys(this.activeWorkflows);
+
+		if (activeWorkflowIds.length === 0) return;
+
+		for (const workflowId of activeWorkflowIds) {
 			await this.remove(workflowId);
 		}
+
+		this.logger.debug('Deactivated all trigger- and poller-based workflows', {
+			workflowIds: activeWorkflowIds,
+		});
 	}
 
 	private async closeTrigger(response: ITriggerResponse, workflowId: string) {