Skip to content

Commit

Permalink
fix(core): Ensure followers do not recover executions from logs (#9785)
Browse files Browse the repository at this point in the history
  • Loading branch information
ivov authored Jun 18, 2024
1 parent 2d02c73 commit 7c358e5
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 3 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ import { EventMessageWorkflow } from '@/eventbus/EventMessageClasses/EventMessag
import type { EventMessageTypes as EventMessage } from '@/eventbus/EventMessageClasses';
import type { WorkflowEntity } from '@/databases/entities/WorkflowEntity';
import { NodeConnectionType } from 'n8n-workflow';
import { OrchestrationService } from '@/services/orchestration.service';
import config from '@/config';

/**
* Workflow producing an execution whose data will be truncated by an instance crash.
Expand Down Expand Up @@ -175,14 +177,25 @@ describe('ExecutionRecoveryService', () => {
let executionRecoveryService: ExecutionRecoveryService;
let push: Push;
let executionRepository: ExecutionRepository;
let orchestrationService: OrchestrationService;

beforeAll(async () => {
await testDb.init();

mockInstance(InternalHooks);
push = mockInstance(Push);
executionRepository = Container.get(ExecutionRepository);
executionRecoveryService = new ExecutionRecoveryService(push, executionRepository);
orchestrationService = Container.get(OrchestrationService);

executionRecoveryService = new ExecutionRecoveryService(
push,
executionRepository,
orchestrationService,
);
});

beforeEach(() => {
config.set('multiMainSetup.instanceType', 'leader');
});

afterEach(async () => {
Expand All @@ -194,7 +207,29 @@ describe('ExecutionRecoveryService', () => {
});

describe('recoverFromLogs', () => {
describe('if no messages', () => {
describe('if follower', () => {
test('should do nothing', async () => {
/**
* Arrange
*/
config.set('multiMainSetup.instanceType', 'follower');
// @ts-expect-error Private method
const amendSpy = jest.spyOn(executionRecoveryService, 'amend');
const messages = setupMessages('123', 'Some workflow');

/**
* Act
*/
await executionRecoveryService.recoverFromLogs('123', messages);

/**
* Assert
*/
expect(amendSpy).not.toHaveBeenCalled();
});
});

describe('if leader, with 0 messages', () => {
test('should return `null` if no execution found', async () => {
/**
* Arrange
Expand Down Expand Up @@ -244,7 +279,7 @@ describe('ExecutionRecoveryService', () => {
});
});

describe('if messages', () => {
describe('if leader, with 1+ messages', () => {
test('should return `null` if no execution found', async () => {
/**
* Arrange
Expand Down
4 changes: 4 additions & 0 deletions packages/cli/src/executions/execution-recovery.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import type { IExecutionResponse } from '@/Interfaces';
import { NodeCrashedError } from '@/errors/node-crashed.error';
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
import { ARTIFICIAL_TASK_DATA } from '@/constants';
import { OrchestrationService } from '@/services/orchestration.service';

/**
* Service for recovering key properties in executions.
Expand All @@ -20,12 +21,15 @@ export class ExecutionRecoveryService {
constructor(
private readonly push: Push,
private readonly executionRepository: ExecutionRepository,
private readonly orchestrationService: OrchestrationService,
) {}

/**
* Recover key properties of a truncated execution using event logs.
*/
async recoverFromLogs(executionId: string, messages: EventMessageTypes[]) {
if (this.orchestrationService.isFollower) return;

const amendedExecution = await this.amend(executionId, messages);

if (!amendedExecution) return null;
Expand Down

0 comments on commit 7c358e5

Please sign in to comment.