fix(core): Ensure followers do not recover executions from logs (#9785)
This commit is contained in:
@@ -20,6 +20,8 @@ import { EventMessageWorkflow } from '@/eventbus/EventMessageClasses/EventMessag
|
|||||||
import type { EventMessageTypes as EventMessage } from '@/eventbus/EventMessageClasses';
|
import type { EventMessageTypes as EventMessage } from '@/eventbus/EventMessageClasses';
|
||||||
import type { WorkflowEntity } from '@/databases/entities/WorkflowEntity';
|
import type { WorkflowEntity } from '@/databases/entities/WorkflowEntity';
|
||||||
import { NodeConnectionType } from 'n8n-workflow';
|
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.
|
* Workflow producing an execution whose data will be truncated by an instance crash.
|
||||||
@@ -175,6 +177,7 @@ describe('ExecutionRecoveryService', () => {
|
|||||||
let executionRecoveryService: ExecutionRecoveryService;
|
let executionRecoveryService: ExecutionRecoveryService;
|
||||||
let push: Push;
|
let push: Push;
|
||||||
let executionRepository: ExecutionRepository;
|
let executionRepository: ExecutionRepository;
|
||||||
|
let orchestrationService: OrchestrationService;
|
||||||
|
|
||||||
beforeAll(async () => {
|
beforeAll(async () => {
|
||||||
await testDb.init();
|
await testDb.init();
|
||||||
@@ -182,7 +185,17 @@ describe('ExecutionRecoveryService', () => {
|
|||||||
mockInstance(InternalHooks);
|
mockInstance(InternalHooks);
|
||||||
push = mockInstance(Push);
|
push = mockInstance(Push);
|
||||||
executionRepository = Container.get(ExecutionRepository);
|
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 () => {
|
afterEach(async () => {
|
||||||
@@ -194,7 +207,29 @@ describe('ExecutionRecoveryService', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe('recoverFromLogs', () => {
|
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 () => {
|
test('should return `null` if no execution found', async () => {
|
||||||
/**
|
/**
|
||||||
* Arrange
|
* Arrange
|
||||||
@@ -244,7 +279,7 @@ describe('ExecutionRecoveryService', () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
describe('if messages', () => {
|
describe('if leader, with 1+ messages', () => {
|
||||||
test('should return `null` if no execution found', async () => {
|
test('should return `null` if no execution found', async () => {
|
||||||
/**
|
/**
|
||||||
* Arrange
|
* Arrange
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import type { IExecutionResponse } from '@/Interfaces';
|
|||||||
import { NodeCrashedError } from '@/errors/node-crashed.error';
|
import { NodeCrashedError } from '@/errors/node-crashed.error';
|
||||||
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
|
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
|
||||||
import { ARTIFICIAL_TASK_DATA } from '@/constants';
|
import { ARTIFICIAL_TASK_DATA } from '@/constants';
|
||||||
|
import { OrchestrationService } from '@/services/orchestration.service';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Service for recovering key properties in executions.
|
* Service for recovering key properties in executions.
|
||||||
@@ -20,12 +21,15 @@ export class ExecutionRecoveryService {
|
|||||||
constructor(
|
constructor(
|
||||||
private readonly push: Push,
|
private readonly push: Push,
|
||||||
private readonly executionRepository: ExecutionRepository,
|
private readonly executionRepository: ExecutionRepository,
|
||||||
|
private readonly orchestrationService: OrchestrationService,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Recover key properties of a truncated execution using event logs.
|
* Recover key properties of a truncated execution using event logs.
|
||||||
*/
|
*/
|
||||||
async recoverFromLogs(executionId: string, messages: EventMessageTypes[]) {
|
async recoverFromLogs(executionId: string, messages: EventMessageTypes[]) {
|
||||||
|
if (this.orchestrationService.isFollower) return;
|
||||||
|
|
||||||
const amendedExecution = await this.amend(executionId, messages);
|
const amendedExecution = await this.amend(executionId, messages);
|
||||||
|
|
||||||
if (!amendedExecution) return null;
|
if (!amendedExecution) return null;
|
||||||
|
|||||||
Reference in New Issue
Block a user