diff --git a/packages/cli/src/__tests__/wait-tracker.test.ts b/packages/cli/src/__tests__/wait-tracker.test.ts index 5f614f42f8..ba23293f7b 100644 --- a/packages/cli/src/__tests__/wait-tracker.test.ts +++ b/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(); const executionRepository = mock(); const multiMainSetup = mock(); - const orchestrationService = new OrchestrationService(mock(), multiMainSetup, mock()); const instanceSettings = mock({ isLeader: true, isMultiMain: false }); const project = mock({ 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({ isLeader: false, isMultiMain: false }), ); diff --git a/packages/cli/src/active-workflow-manager.ts b/packages/cli/src/active-workflow-manager.ts index 58d78c63a4..4c94f16a7b 100644 --- a/packages/cli/src/active-workflow-manager.ts +++ b/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(); } diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index f10e06e89b..ea9b18ee24 100644 --- a/packages/cli/src/commands/start.ts +++ b/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() { diff --git a/packages/cli/src/license.ts b/packages/cli/src/license.ts index c5583f091c..9e192d2f6e 100644 --- a/packages/cli/src/license.ts +++ b/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(); } diff --git a/packages/cli/src/modules/insights/__tests__/insights-collection.service.test.ts b/packages/cli/src/modules/insights/__tests__/insights-collection.service.test.ts index e20e5ecb7e..fb033a63a2 100644 --- a/packages/cli/src/modules/insights/__tests__/insights-collection.service.test.ts +++ b/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(), + mockLogger(), ); }); diff --git a/packages/cli/src/modules/insights/__tests__/insights.service.test.ts b/packages/cli/src/modules/insights/__tests__/insights.service.test.ts index df4487a449..78abaf4060 100644 --- a/packages/cli/src/modules/insights/__tests__/insights.service.test.ts +++ b/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(), mock(), licenseMock, - mock(), ); }); @@ -603,7 +601,6 @@ describe('getMaxAgeInDaysAndGranularity', () => { mock(), mock(), licenseMock, - mock(), ); }); diff --git a/packages/cli/src/modules/insights/insights-collection.service.ts b/packages/cli/src/modules/insights/insights-collection.service.ts index 0762760ee8..a436f86da1 100644 --- a/packages/cli/src/modules/insights/insights-collection.service.ts +++ b/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'); } } diff --git a/packages/cli/src/modules/insights/insights-compaction.service.ts b/packages/cli/src/modules/insights/insights-compaction.service.ts index e53f381ea2..47c852d864 100644 --- a/packages/cli/src/modules/insights/insights-compaction.service.ts +++ b/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'); } } diff --git a/packages/cli/src/modules/insights/insights.service.ts b/packages/cli/src/modules/insights/insights.service.ts index cc2466c06c..58d9924312 100644 --- a/packages/cli/src/modules/insights/insights.service.ts +++ b/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() diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index cbabd68504..4a3640f991 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -81,7 +81,6 @@ describe('ScalingService', () => { mock(), instanceSettings, mock(), - mock(), ); getRunningJobsCountSpy = jest.spyOn(scalingService, 'getRunningJobsCount'); diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index ac14b11655..0236d96c79 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/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'); diff --git a/packages/cli/src/services/pruning/__tests__/pruning.service.test.ts b/packages/cli/src/services/pruning/__tests__/pruning.service.test.ts index 050d1b82d9..9b3d55a81e 100644 --- a/packages/cli/src/services/pruning/__tests__/pruning.service.test.ts +++ b/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({ isLeader: true, isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock(), ); const startPruningSpy = jest.spyOn(pruningService, 'startPruning'); @@ -38,9 +33,6 @@ describe('PruningService', () => { mock({ isLeader: false, isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), 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({ isLeader: true, isMultiMain: true }), - mock(), - mock(), - mock({ - multiMainSetup: mock({ 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({ isLeader: true, instanceType: 'main', isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: true }), ); @@ -100,9 +62,6 @@ describe('PruningService', () => { mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: false }), ); @@ -115,9 +74,6 @@ describe('PruningService', () => { mock({ isLeader: false, instanceType: 'worker', isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: true }), ); @@ -135,9 +91,6 @@ describe('PruningService', () => { }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: true }), ); @@ -152,9 +105,6 @@ describe('PruningService', () => { mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: false }), ); @@ -179,9 +129,6 @@ describe('PruningService', () => { mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), mock(), mock(), - mock({ - multiMainSetup: mock(), - }), mock({ pruneData: true }), ); diff --git a/packages/cli/src/services/pruning/pruning.service.ts b/packages/cli/src/services/pruning/pruning.service.ts index ebac8ec462..171b606fab 100644 --- a/packages/cli/src/services/pruning/pruning.service.ts +++ b/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) { diff --git a/packages/cli/src/wait-tracker.ts b/packages/cli/src/wait-tracker.ts index cca69548bd..848dc919b1 100644 --- a/packages/cli/src/wait-tracker.ts +++ b/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'); } } diff --git a/packages/cli/test/integration/pruning.service.test.ts b/packages/cli/test/integration/pruning.service.test.ts index b2d84ea362..0e7fc05062 100644 --- a/packages/cli/test/integration/pruning.service.test.ts +++ b/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, ); diff --git a/packages/core/src/execution-engine/active-workflows.ts b/packages/core/src/execution-engine/active-workflows.ts index 1c1b5e2d14..b62f49628b 100644 --- a/packages/core/src/execution-engine/active-workflows.ts +++ b/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) {