move leader logic to insights service. refactor tests

Author: guillaumejacquart

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

@@ -5,8 +5,10 @@ import type { WorkflowEntity } from '@n8n/db';
 import type { IWorkflowDb } from '@n8n/db';
 import type { WorkflowExecuteAfterContext } from '@n8n/decorators';
 import { Container } from '@n8n/di';
+import type { MockProxy } from 'jest-mock-extended';
 import { mock } from 'jest-mock-extended';
 import { DateTime } from 'luxon';
+import type { InstanceSettings, InstanceType } from 'n8n-core';
 import type { IRun } from 'n8n-workflow';
 
 import { mockLogger } from '@test/mocking';
@@ -52,17 +54,30 @@ describe('startTimers', () => {
 	let compactionService: InsightsCompactionService;
 	let collectionService: InsightsCollectionService;
 	let pruningService: InsightsPruningService;
+	let instanceSettings: MockProxy<InstanceSettings>;
+	let isLeader = jest.fn(() => true);
+	let isPruningEnabled = jest.fn(() => false);
 
 	beforeAll(() => {
 		compactionService = mock<InsightsCompactionService>();
 		collectionService = mock<InsightsCollectionService>();
-		pruningService = mock<InsightsPruningService>({ isPruningEnabled: true });
+		pruningService = mock<InsightsPruningService>();
+		Object.defineProperty(pruningService, 'isPruningEnabled', {
+			get: isPruningEnabled,
+		});
+		instanceSettings = mock<InstanceSettings>({
+			instanceType: 'main',
+		});
+		Object.defineProperty(instanceSettings, 'isLeader', {
+			get: isLeader,
+		});
 		insightsService = new InsightsService(
 			mock<InsightsByPeriodRepository>(),
 			compactionService,
 			collectionService,
 			pruningService,
 			mock<LicenseState>(),
+			instanceSettings,
 			mockLogger(),
 		);
 	});
@@ -71,21 +86,48 @@ describe('startTimers', () => {
 		jest.clearAllMocks();
 	});
 
-	test('starts compaction, flushing and pruning timers', () => {
+	test('starts compaction, flushing timers for main leader instance', () => {
+		isLeader.mockReturnValueOnce(true);
+		isPruningEnabled.mockReturnValueOnce(false);
+		insightsService.startTimers();
+
+		expect(collectionService.startFlushingTimer).toHaveBeenCalled();
+		expect(compactionService.startCompactionTimer).toHaveBeenCalled();
+		expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
+	});
+
+	test('starts compaction, flushing and pruning timers for main leader instance with pruning enabled', () => {
+		isLeader.mockReturnValueOnce(true);
+		isPruningEnabled.mockReturnValueOnce(true);
 		insightsService.startTimers();
 
 		expect(collectionService.startFlushingTimer).toHaveBeenCalled();
 		expect(compactionService.startCompactionTimer).toHaveBeenCalled();
 		expect(pruningService.startPruningTimer).toHaveBeenCalled();
 	});
 
-	test('starts only collection flushing timer when onlyCollection is true', () => {
-		insightsService.startTimers({ onlyCollection: true });
+	test('starts only collection flushing timer for webhook instance', () => {
+		(instanceSettings as any).instanceType = 'webhook';
+		insightsService.startTimers();
 
 		expect(collectionService.startFlushingTimer).toHaveBeenCalled();
 		expect(compactionService.startCompactionTimer).not.toHaveBeenCalled();
 		expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
 	});
+
+	test.each<InstanceType>(['worker', 'main'])(
+		'do not start any timers for instance of type %s when not leader',
+		(instanceType: InstanceType) => {
+			(instanceSettings as any).instanceType = instanceType;
+			isLeader.mockReturnValue(false);
+
+			insightsService.startTimers();
+
+			expect(collectionService.startFlushingTimer).not.toHaveBeenCalled();
+			expect(compactionService.startCompactionTimer).not.toHaveBeenCalled();
+			expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
+		},
+	);
 });
 
 describe('getInsightsSummary', () => {
@@ -553,6 +595,7 @@ describe('getAvailableDateRanges', () => {
 			mock<InsightsCollectionService>(),
 			mock<InsightsPruningService>(),
 			licenseMock,
+			mock<InstanceSettings>(),
 			mockLogger(),
 		);
 	});
@@ -655,6 +698,7 @@ describe('getMaxAgeInDaysAndGranularity', () => {
 			mock<InsightsCollectionService>(),
 			mock<InsightsPruningService>(),
 			licenseMock,
+			mock<InstanceSettings>(),
 			mockLogger(),
 		);
 	});
@@ -743,6 +787,7 @@ describe('shutdown', () => {
 			mockCollectionService,
 			mockPruningService,
 			mock<LicenseState>(),
+			mock<InstanceSettings>(),
 			mockLogger(),
 		);
 	});
@@ -758,74 +803,6 @@ describe('shutdown', () => {
 	});
 });
 
-describe('timers', () => {
-	let insightsService: InsightsService;
-
-	const mockCollectionService = mock<InsightsCollectionService>({
-		startFlushingTimer: jest.fn(),
-		stopFlushingTimer: jest.fn(),
-	});
-
-	const mockCompactionService = mock<InsightsCompactionService>({
-		startCompactionTimer: jest.fn(),
-		stopCompactionTimer: jest.fn(),
-	});
-
-	const mockPruningService = mock<InsightsPruningService>({
-		startPruningTimer: jest.fn(),
-		stopPruningTimer: jest.fn(),
-		isPruningEnabled: false,
-	});
-
-	const mockedLogger = mockLogger();
-	const mockedConfig = mock<InsightsConfig>({
-		maxAgeDays: -1,
-	});
-
-	beforeAll(() => {
-		insightsService = new InsightsService(
-			mock<InsightsByPeriodRepository>(),
-			mockCompactionService,
-			mockCollectionService,
-			mockPruningService,
-			mock<LicenseState>(),
-			mockedLogger,
-		);
-	});
-
-	test('startTimers starts timers except pruning', () => {
-		// ACT
-		insightsService.startTimers();
-
-		// ASSERT
-		expect(mockCompactionService.startCompactionTimer).toHaveBeenCalled();
-		expect(mockCollectionService.startFlushingTimer).toHaveBeenCalled();
-		expect(mockPruningService.startPruningTimer).not.toHaveBeenCalled();
-	});
-
-	test('startTimers starts pruning timer', () => {
-		// ARRANGE
-		mockedConfig.maxAgeDays = 30;
-		Object.defineProperty(mockPruningService, 'isPruningEnabled', { value: true });
-
-		// ACT
-		insightsService.startTimers();
-
-		// ASSERT
-		expect(mockPruningService.startPruningTimer).toHaveBeenCalled();
-	});
-
-	test('stopTimers stops timers', () => {
-		// ACT
-		insightsService.stopTimers();
-
-		// ASSERT
-		expect(mockCompactionService.stopCompactionTimer).toHaveBeenCalled();
-		expect(mockCollectionService.stopFlushingTimer).toHaveBeenCalled();
-		expect(mockPruningService.stopPruningTimer).toHaveBeenCalled();
-	});
-});
-
 describe('legacy sqlite (without pooling) handles concurrent insights db process without throwing', () => {
 	let initialFlushBatchSize: number;
 	let insightsConfig: InsightsConfig;

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

@@ -76,7 +76,6 @@ export class InsightsCollectionService {
 	startFlushingTimer() {
 		this.isAsynchronouslySavingInsights = true;
 		this.scheduleFlushing();
-		this.logger.debug('Started flushing timer');
 	}
 
 	scheduleFlushing() {

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

@@ -1,40 +1,15 @@
-import { Logger } from '@n8n/backend-common';
 import type { BaseN8nModule } from '@n8n/decorators';
-import { N8nModule, OnLeaderStepdown, OnLeaderTakeover } from '@n8n/decorators';
-import { InstanceSettings } from 'n8n-core';
+import { N8nModule } from '@n8n/decorators';
 
 import { InsightsService } from './insights.service';
 
 import './insights.controller';
 
 @N8nModule()
 export class InsightsModule implements BaseN8nModule {
-	constructor(
-		private readonly logger: Logger,
-		private readonly insightsService: InsightsService,
-		private readonly instanceSettings: InstanceSettings,
-	) {
-		this.logger = this.logger.scoped('insights');
-	}
+	constructor(private readonly insightsService: InsightsService) {}
 
 	initialize() {
-		// We want to initialize the insights background process (schedulers) for the main leader instance
-		if (this.instanceSettings.isLeader) {
-			this.insightsService.startTimers();
-		} else if (this.instanceSettings.instanceType === 'webhook') {
-			// Webhook instances collect insights data independently
-			// So we only start the collection timers, compaction and pruning are done by the main instance
-			this.insightsService.startTimers({ onlyCollection: true });
-		}
-	}
-
-	@OnLeaderTakeover()
-	startBackgroundProcess() {
 		this.insightsService.startTimers();
 	}
-
-	@OnLeaderStepdown()
-	stopBackgroundProcess() {
-		this.insightsService.stopTimers();
-	}
 }

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

@@ -4,8 +4,9 @@ import {
 	INSIGHTS_DATE_RANGE_KEYS,
 } from '@n8n/api-types';
 import { LicenseState, Logger } from '@n8n/backend-common';
-import { OnShutdown } from '@n8n/decorators';
+import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
+import { InstanceSettings } from 'n8n-core';
 import { UserError } from 'n8n-workflow';
 
 import type { PeriodUnit, TypeUnit } from './database/entities/insights-shared';
@@ -33,26 +34,33 @@ export class InsightsService {
 		private readonly collectionService: InsightsCollectionService,
 		private readonly pruningService: InsightsPruningService,
 		private readonly licenseState: LicenseState,
+		private readonly instanceSettings: InstanceSettings,
 		private readonly logger: Logger,
 	) {
 		this.logger = this.logger.scoped('insights');
 	}
 
-	startTimers({ onlyCollection } = { onlyCollection: false }) {
+	@OnLeaderTakeover()
+	startTimers() {
+		if (!this.instanceSettings.isLeader && this.instanceSettings.instanceType !== 'webhook') {
+			return;
+		}
+
 		this.collectionService.startFlushingTimer();
-		if (!onlyCollection) {
+		this.logger.debug('Started collection flushing scheduler');
+
+		// Start compaction and pruning timers for main leader instance only
+		if (this.instanceSettings.instanceType !== 'webhook') {
 			this.compactionService.startCompactionTimer();
+			this.logger.debug('Started compaction scheduler');
 			if (this.pruningService.isPruningEnabled) {
 				this.pruningService.startPruningTimer();
+				this.logger.debug('Started pruning scheduler');
 			}
 		}
-		this.logger.debug(
-			onlyCollection
-				? 'Starting collection flushing schedulers'
-				: 'Starting compaction, flushing and pruning schedulers',
-		);
 	}
 
+	@OnLeaderStepdown()
 	stopTimers() {
 		this.compactionService.stopCompactionTimer();
 		this.collectionService.stopFlushingTimer();