mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-03-05 20:50:17 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(core): Make PruningService.init and WaitTracker.init consistent (no-changelog) (#9761)

Browse source
This commit is contained in:
Iván Ovejero 2024-06-17 12:49:40 +02:00 committed by GitHub
parent 7c70b782a1
commit f7352b6a8f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 43 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
packages/cli/src/WaitTracker.ts
View file

@ -30,19 +30,19 @@ export class WaitTracker {
private readonly orchestrationService: OrchestrationService,
) {}
/**
* @important Requires `OrchestrationService` to be initialized.
*/
init() {
const { isSingleMainSetup, isLeader, multiMainSetup } = this.orchestrationService;
if (isSingleMainSetup) {
this.startTracking();
return;
}
const { isLeader, isMultiMainSetupEnabled } = this.orchestrationService;
if (isLeader) this.startTracking();
multiMainSetup
.on('leader-takeover', () => this.startTracking())
.on('leader-stepdown', () => this.stopTracking());
if (isMultiMainSetupEnabled) {
this.orchestrationService.multiMainSetup
.on('leader-takeover', () => this.startTracking())
.on('leader-stepdown', () => this.stopTracking());
}
}
private startTracking() {

25
packages/cli/src/commands/start.ts
View file

@ -199,7 +199,10 @@ export class Start extends BaseCommand {
}
async initOrchestration() {
if (config.getEnv('executions.mode') !== 'queue') return;
if (config.getEnv('executions.mode') === 'regular') {
config.set('multiMainSetup.instanceType', 'leader');
return;
}
if (
config.getEnv('multiMainSetup.enabled') &&
@ -290,7 +293,7 @@ export class Start extends BaseCommand {
await this.server.start();
await this.initPruning();
Container.get(PruningService).init();
if (config.getEnv('executions.mode') === 'regular') {
await this.runEnqueuedExecutions();
@ -333,24 +336,6 @@ export class Start extends BaseCommand {
}
}
async initPruning() {
this.pruningService = Container.get(PruningService);
this.pruningService.startPruning();
if (config.getEnv('executions.mode') !== 'queue') return;
const orchestrationService = Container.get(OrchestrationService);
await orchestrationService.init();
if (!orchestrationService.isMultiMainSetupEnabled) return;
orchestrationService.multiMainSetup
.on('leader-stepdown', () => this.pruningService.stopPruning())
.on('leader-takeover', () => this.pruningService.startPruning());
}
async catch(error: Error) {
if (error.stack) this.logger.error(error.stack);
await this.exitWithCrash('Exiting due to an error.', error);

16
packages/cli/src/services/pruning.service.ts
View file

@ -6,6 +6,7 @@ import { ExecutionRepository } from '@db/repositories/execution.repository';
import { Logger } from '@/Logger';
import { jsonStringify } from 'n8n-workflow';
import { OnShutdown } from '@/decorators/OnShutdown';
import { OrchestrationService } from './orchestration.service';
@Service()
export class PruningService {
@ -26,8 +27,23 @@ export class PruningService {
private readonly logger: Logger,
private readonly executionRepository: ExecutionRepository,
private readonly binaryDataService: BinaryDataService,
private readonly orchestrationService: OrchestrationService,
) {}
/**
* @important Requires `OrchestrationService` to be initialized.
*/
init() {
const { isLeader, isMultiMainSetupEnabled } = this.orchestrationService;
if (isLeader) this.startPruning();
if (isMultiMainSetupEnabled) {
this.orchestrationService.multiMainSetup.on('leader-takeover', () => this.startPruning());
this.orchestrationService.multiMainSetup.on('leader-stepdown', () => this.stopPruning());
}
}
private isPruningEnabled() {
if (
!config.getEnv('executions.pruneData') ||

3
packages/cli/test/integration/pruning.service.test.ts
View file

@ -14,6 +14,8 @@ import { Logger } from '@/Logger';
import { mockInstance } from '../shared/mocking';
import { createWorkflow } from './shared/db/workflows';
import { createExecution, createSuccessfulExecution } from './shared/db/executions';
import { mock } from 'jest-mock-extended';
import type { OrchestrationService } from '@/services/orchestration.service';
describe('softDeleteOnPruningCycle()', () => {
let pruningService: PruningService;
@ -29,6 +31,7 @@ describe('softDeleteOnPruningCycle()', () => {
mockInstance(Logger),
Container.get(ExecutionRepository),
mockInstance(BinaryDataService),
mock<OrchestrationService>(),
);
workflow = await createWorkflow();

11
packages/cli/test/unit/WaitTracker.test.ts
View file

@ -34,7 +34,8 @@ describe('WaitTracker', () => {
});
describe('init()', () => {
it('should query DB for waiting executions', async () => {
it('should query DB for waiting executions if leader', async () => {
jest.spyOn(orchestrationService, 'isLeader', 'get').mockReturnValue(true);
executionRepository.getWaitingExecutions.mockResolvedValue([execution]);
waitTracker.init();
@ -42,6 +43,14 @@ describe('WaitTracker', () => {
expect(executionRepository.getWaitingExecutions).toHaveBeenCalledTimes(1);
});
it('if follower, should do nothing', () => {
executionRepository.getWaitingExecutions.mockResolvedValue([]);
waitTracker.init();
expect(executionRepository.findSingleExecution).not.toHaveBeenCalled();
});
it('if no executions to start, should do nothing', () => {
executionRepository.getWaitingExecutions.mockResolvedValue([]);