mirror of
https://github.com/n8n-io/n8n.git
synced 2024-12-25 20:54:07 -08:00
refactor(core): Don't use DB transactions on ExecutionRepository.createNewExecution (#8002)
Saving execution data is one of the slowest DB operations in the application, and is likely behind some of the sqlite transaction concurrency issues we've been seeing. This not only remove the 2 separate transactions for saving `ExecutionEntity` and `ExecutionData`, but also remove fields from `ExecutionData.workflowData` that don't need to be saved (like `tags`, `shared`, `statistics`, `triggerCount`, etc).
This commit is contained in:
parent
19e88ec8a1
commit
1d870412ca
|
@ -60,9 +60,7 @@ export class ActiveExecutions {
|
||||||
fullExecutionData.workflowId = workflowId;
|
fullExecutionData.workflowId = workflowId;
|
||||||
}
|
}
|
||||||
|
|
||||||
const executionResult =
|
executionId = await Container.get(ExecutionRepository).createNewExecution(fullExecutionData);
|
||||||
await Container.get(ExecutionRepository).createNewExecution(fullExecutionData);
|
|
||||||
executionId = executionResult.id;
|
|
||||||
if (executionId === undefined) {
|
if (executionId === undefined) {
|
||||||
throw new ApplicationError('There was an issue assigning an execution id to the execution');
|
throw new ApplicationError('There was an issue assigning an execution id to the execution');
|
||||||
}
|
}
|
||||||
|
|
|
@ -213,17 +213,17 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
|
||||||
return rest;
|
return rest;
|
||||||
}
|
}
|
||||||
|
|
||||||
async createNewExecution(execution: ExecutionPayload) {
|
async createNewExecution(execution: ExecutionPayload): Promise<string> {
|
||||||
const { data, workflowData, ...rest } = execution;
|
const { data, workflowData, ...rest } = execution;
|
||||||
|
const { identifiers: inserted } = await this.insert(rest);
|
||||||
const newExecution = await this.save(rest);
|
const { id: executionId } = inserted[0] as { id: string };
|
||||||
await this.executionDataRepository.save({
|
const { connections, nodes, name } = workflowData ?? {};
|
||||||
execution: newExecution,
|
await this.executionDataRepository.insert({
|
||||||
workflowData,
|
executionId,
|
||||||
|
workflowData: { connections, nodes, name },
|
||||||
data: stringify(data),
|
data: stringify(data),
|
||||||
});
|
});
|
||||||
|
return String(executionId);
|
||||||
return newExecution;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async markAsCrashed(executionIds: string[]) {
|
async markAsCrashed(executionIds: string[]) {
|
||||||
|
|
|
@ -0,0 +1,53 @@
|
||||||
|
import Container from 'typedi';
|
||||||
|
import { ExecutionRepository } from '@db/repositories/execution.repository';
|
||||||
|
import { ExecutionDataRepository } from '@db/repositories/executionData.repository';
|
||||||
|
import * as testDb from '../../shared/testDb';
|
||||||
|
import { createWorkflow } from '../../shared/db/workflows';
|
||||||
|
|
||||||
|
describe('ExecutionRepository', () => {
|
||||||
|
beforeAll(async () => {
|
||||||
|
await testDb.init();
|
||||||
|
});
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
await testDb.truncate(['Workflow', 'Execution']);
|
||||||
|
});
|
||||||
|
|
||||||
|
afterAll(async () => {
|
||||||
|
await testDb.terminate();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('createNewExecution', () => {
|
||||||
|
it('should save execution data', async () => {
|
||||||
|
const executionRepo = Container.get(ExecutionRepository);
|
||||||
|
const workflow = await createWorkflow();
|
||||||
|
const executionId = await executionRepo.createNewExecution({
|
||||||
|
workflowId: workflow.id,
|
||||||
|
data: {
|
||||||
|
resultData: {},
|
||||||
|
},
|
||||||
|
workflowData: workflow,
|
||||||
|
mode: 'manual',
|
||||||
|
startedAt: new Date(),
|
||||||
|
status: 'new',
|
||||||
|
finished: false,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(executionId).toBeDefined();
|
||||||
|
|
||||||
|
const executionEntity = await executionRepo.findOneBy({ id: executionId });
|
||||||
|
expect(executionEntity?.id).toEqual(executionId);
|
||||||
|
expect(executionEntity?.workflowId).toEqual(workflow.id);
|
||||||
|
expect(executionEntity?.status).toEqual('new');
|
||||||
|
|
||||||
|
const executionDataRepo = Container.get(ExecutionDataRepository);
|
||||||
|
const executionData = await executionDataRepo.findOneBy({ executionId });
|
||||||
|
expect(executionData?.workflowData).toEqual({
|
||||||
|
connections: workflow.connections,
|
||||||
|
nodes: workflow.nodes,
|
||||||
|
name: workflow.name,
|
||||||
|
});
|
||||||
|
expect(executionData?.data).toEqual('[{"resultData":"1"},{}]');
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
|
@ -12,9 +12,7 @@ const FAKE_EXECUTION_ID = '15';
|
||||||
const FAKE_SECOND_EXECUTION_ID = '20';
|
const FAKE_SECOND_EXECUTION_ID = '20';
|
||||||
|
|
||||||
const updateExistingExecution = jest.fn();
|
const updateExistingExecution = jest.fn();
|
||||||
const createNewExecution = jest.fn(async () => {
|
const createNewExecution = jest.fn(async () => FAKE_EXECUTION_ID);
|
||||||
return { id: FAKE_EXECUTION_ID };
|
|
||||||
});
|
|
||||||
|
|
||||||
Container.set(ExecutionRepository, {
|
Container.set(ExecutionRepository, {
|
||||||
updateExistingExecution,
|
updateExistingExecution,
|
||||||
|
|
Loading…
Reference in a new issue