mirror of
https://github.com/n8n-io/n8n.git
synced 2026-08-30 18:01:23 +08:00
Ali Elkhateeb
438 lines
15 KiB
TypeScript
438 lines
15 KiB
TypeScript
import { createWorkflow, testDb } from '@n8n/backend-test-utils';
|
|
import { ExecutionDataRepository, ExecutionRepository } from '@n8n/db';
|
|
import type { IExecutionResponse } from '@n8n/db';
|
|
import { Container } from '@n8n/di';
|
|
import { createEmptyRunExecutionData } from 'n8n-workflow';
|
|
|
|
import { DbStore } from '@/executions/execution-data/db-store';
|
|
import { MissingExecutionDataError } from '@/executions/execution-data/missing-execution-data.error';
|
|
import { ExecutionPersistence } from '@/executions/execution-persistence';
|
|
|
|
describe('ExecutionPersistence', () => {
|
|
beforeAll(async () => {
|
|
await testDb.init();
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
await testDb.truncate(['WorkflowEntity', 'ExecutionEntity']);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await testDb.terminate();
|
|
});
|
|
|
|
describe('create', () => {
|
|
it('should save execution and execution data to database', async () => {
|
|
const executionRepo = Container.get(ExecutionRepository);
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data: {
|
|
// @ts-expect-error Partial data for test
|
|
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');
|
|
expect(executionEntity?.storedAt).toEqual('db');
|
|
|
|
const executionDataRepository = Container.get(ExecutionDataRepository);
|
|
const executionData = await executionDataRepository.findOneBy({ executionId });
|
|
expect(executionData?.workflowData).toEqual({
|
|
id: workflow.id,
|
|
connections: workflow.connections,
|
|
nodes: workflow.nodes,
|
|
name: workflow.name,
|
|
settings: workflow.settings,
|
|
nodeGroups: workflow.nodeGroups,
|
|
});
|
|
expect(executionData?.data).toEqual('[{"resultData":"1"},{}]');
|
|
});
|
|
});
|
|
|
|
describe('storedAt CHECK constraint (AllowAzureStoredAt migration)', () => {
|
|
it("allows persisting an execution with storedAt 'az' (legacy 3-value check was widened, not AND-ed)", async () => {
|
|
const executionRepo = Container.get(ExecutionRepository);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
// Inserting 'az' would throw a CHECK violation if the original
|
|
// `IN ('db','fs','s3')` constraint were still present.
|
|
const result = await executionRepo.insert({
|
|
workflowId: workflow.id,
|
|
status: 'new',
|
|
finished: false,
|
|
mode: 'manual',
|
|
createdAt: new Date(),
|
|
storedAt: 'az',
|
|
});
|
|
|
|
const id = String(result.identifiers[0].id);
|
|
const found = await executionRepo.findOneBy({ id });
|
|
expect(found?.storedAt).toEqual('az');
|
|
});
|
|
});
|
|
|
|
describe('updateExistingExecution (db overwrite path)', () => {
|
|
it('should preserve the original workflowVersionId when overwriting data and workflowData', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const executionDataRepository = Container.get(ExecutionDataRepository);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data: createEmptyRunExecutionData(),
|
|
workflowData: { ...workflow, versionId: 'v-original' },
|
|
mode: 'manual',
|
|
status: 'new',
|
|
finished: false,
|
|
});
|
|
|
|
const updatedData = createEmptyRunExecutionData();
|
|
updatedData.resultData.lastNodeExecuted = 'NodeX';
|
|
|
|
await executionPersistence.updateExistingExecution(executionId, {
|
|
data: updatedData,
|
|
workflowData: { ...workflow, versionId: 'v-different' },
|
|
status: 'success',
|
|
});
|
|
|
|
const executionData = await executionDataRepository.findOneBy({ executionId });
|
|
expect(executionData?.workflowVersionId).toEqual('v-original');
|
|
expect(executionData?.data).toContain('NodeX');
|
|
});
|
|
|
|
it('should roll back the status change when the data row is missing during an overwrite', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const executionRepo = Container.get(ExecutionRepository);
|
|
const executionDataRepository = Container.get(ExecutionDataRepository);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data: createEmptyRunExecutionData(),
|
|
workflowData: workflow,
|
|
mode: 'manual',
|
|
status: 'new',
|
|
finished: false,
|
|
});
|
|
|
|
await executionDataRepository.delete({ executionId });
|
|
|
|
await expect(
|
|
executionPersistence.updateExistingExecution(executionId, {
|
|
data: createEmptyRunExecutionData(),
|
|
workflowData: workflow,
|
|
status: 'success',
|
|
}),
|
|
).rejects.toThrow(MissingExecutionDataError);
|
|
|
|
const executionEntity = await executionRepo.findOneBy({ id: executionId });
|
|
expect(executionEntity?.status).toEqual('new');
|
|
});
|
|
|
|
// CAT-3862: `ActiveExecutions.add` claims an existing execution through this
|
|
// path. It used to require `waiting` unconditionally, so an execution enqueued
|
|
// before a restart (status `new`) could never be claimed and stayed `new` forever.
|
|
describe.each([
|
|
{ requireStatus: 'new' as const, expected: true },
|
|
{ requireStatus: 'waiting' as const, expected: false },
|
|
])(
|
|
'claiming a `new` execution with requireStatus: $requireStatus',
|
|
({ requireStatus, expected }) => {
|
|
it(`returns ${expected}`, async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const executionRepo = Container.get(ExecutionRepository);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data: createEmptyRunExecutionData(),
|
|
workflowData: workflow,
|
|
mode: 'webhook',
|
|
status: 'new',
|
|
finished: false,
|
|
});
|
|
|
|
const claimed = await executionPersistence.updateExistingExecution(
|
|
executionId,
|
|
{ data: createEmptyRunExecutionData(), waitTill: null, status: 'running' },
|
|
{ requireStatus },
|
|
);
|
|
|
|
expect(claimed).toBe(expected);
|
|
const row = await executionRepo.findOneBy({ id: executionId });
|
|
expect(row?.status).toEqual(expected ? 'running' : 'new');
|
|
});
|
|
},
|
|
);
|
|
});
|
|
|
|
describe('findSingleExecution', () => {
|
|
it('returns the persisted bundle byte size and the workflow version id', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data: createEmptyRunExecutionData(),
|
|
workflowData: {
|
|
...workflow,
|
|
id: 'wf-1',
|
|
name: 'wf',
|
|
nodes: [],
|
|
connections: {},
|
|
settings: {},
|
|
versionId: 'v-roundtrip-456',
|
|
},
|
|
mode: 'manual',
|
|
status: 'new',
|
|
finished: false,
|
|
});
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
});
|
|
|
|
// 51 (run data) + 83 (workflow snapshot incl. empty nodeGroups) + 15 ("v-roundtrip-456") = 149 bytes
|
|
expect(execution?.jsonSizeBytes).toBe(149);
|
|
expect(execution?.workflowVersionId).toBe('v-roundtrip-456');
|
|
});
|
|
|
|
it('persists and reads back binaryDataSizeBytes from offloaded binary in the run data', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
|
|
const data = createEmptyRunExecutionData();
|
|
data.resultData.runData = {
|
|
Node: [
|
|
{
|
|
data: {
|
|
main: [
|
|
[
|
|
{
|
|
json: {},
|
|
binary: {
|
|
file: {
|
|
data: '',
|
|
mimeType: 'application/octet-stream',
|
|
id: 'filesystem-v2:abc',
|
|
bytes: 100,
|
|
},
|
|
},
|
|
},
|
|
],
|
|
],
|
|
},
|
|
},
|
|
],
|
|
} as unknown as typeof data.resultData.runData;
|
|
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data,
|
|
workflowData: workflow,
|
|
mode: 'manual',
|
|
status: 'new',
|
|
finished: false,
|
|
});
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
});
|
|
|
|
expect(execution?.binaryDataSizeBytes).toBe(100);
|
|
});
|
|
});
|
|
|
|
describe('display size guard (maxDataSizeBytes)', () => {
|
|
const ONE_MB = 1024 * 1024;
|
|
|
|
/** Create a successful execution whose run data embeds a string of roughly `sizeBytes`. */
|
|
const createSizedExecution = async (sizeBytes: number) => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } });
|
|
const data = createEmptyRunExecutionData();
|
|
data.resultData.runData = {
|
|
bigNode: [
|
|
{
|
|
startTime: 0,
|
|
executionTime: 1,
|
|
executionIndex: 0,
|
|
source: [],
|
|
data: { main: [[{ json: { big: 'x'.repeat(sizeBytes) } }]] },
|
|
},
|
|
],
|
|
};
|
|
const executionId = await executionPersistence.create({
|
|
workflowId: workflow.id,
|
|
data,
|
|
workflowData: workflow,
|
|
mode: 'manual',
|
|
status: 'success',
|
|
finished: true,
|
|
});
|
|
return { workflow, executionId };
|
|
};
|
|
|
|
it('findSingleExecution omits oversized data and flags it (without parsing)', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { workflow, executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
maxDataSizeBytes: ONE_MB,
|
|
});
|
|
|
|
expect(execution?.dataTooLargeToDisplay).toBe(true);
|
|
expect(execution?.jsonSizeBytes).toBeGreaterThan(ONE_MB);
|
|
expect(execution?.data?.resultData?.runData).toEqual({});
|
|
// workflow snapshot still loaded (DB store), not the empty stub
|
|
expect(execution?.workflowData?.id).toBe(workflow.id);
|
|
expect(execution?.workflowData?.name).toBe(workflow.name);
|
|
});
|
|
|
|
it('findSingleExecution returns data when under the threshold', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { executionId } = await createSizedExecution(1024); // ~1 KB
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
maxDataSizeBytes: ONE_MB,
|
|
});
|
|
|
|
expect(execution?.dataTooLargeToDisplay).toBeFalsy();
|
|
expect(execution?.data?.resultData?.runData).toHaveProperty('bigNode');
|
|
});
|
|
|
|
it('findSingleExecution loads unconditionally when no threshold is given', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
});
|
|
|
|
expect(execution?.dataTooLargeToDisplay).toBeFalsy();
|
|
expect(execution?.data?.resultData?.runData).toHaveProperty('bigNode');
|
|
});
|
|
|
|
it('findMultipleExecutions omits oversized data and flags it', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { workflow } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const executions = await executionPersistence.findMultipleExecutions(
|
|
{ where: { workflowId: workflow.id, status: 'success' }, order: { id: 'DESC' }, take: 1 },
|
|
{ includeData: true, unflattenData: true, maxDataSizeBytes: ONE_MB },
|
|
);
|
|
|
|
expect(executions[0]?.dataTooLargeToDisplay).toBe(true);
|
|
expect(executions[0]?.data?.resultData?.runData).toEqual({});
|
|
});
|
|
|
|
it('findMultipleExecutions flags legacy oversized rows after reading (jsonSizeBytes unknown)', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const executionRepository = Container.get(ExecutionRepository);
|
|
const { workflow, executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
// Legacy row: size unknown, so it's read and decided by raw byte length.
|
|
await executionRepository.update({ id: executionId }, { jsonSizeBytes: 0 });
|
|
|
|
const executions = await executionPersistence.findMultipleExecutions(
|
|
{ where: { workflowId: workflow.id, status: 'success' }, order: { id: 'DESC' }, take: 1 },
|
|
{ includeData: true, unflattenData: true, maxDataSizeBytes: ONE_MB },
|
|
);
|
|
|
|
expect(executions[0]?.dataTooLargeToDisplay).toBe(true);
|
|
expect(executions[0]?.data?.resultData?.runData).toEqual({});
|
|
});
|
|
|
|
it('checks the DB size and skips the read for legacy rows (jsonSizeBytes unknown)', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const executionRepository = Container.get(ExecutionRepository);
|
|
const dbStore = Container.get(DbStore);
|
|
const { executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
// Simulate a row written before jsonSizeBytes existed: 0 means unknown.
|
|
await executionRepository.update({ id: executionId }, { jsonSizeBytes: 0 });
|
|
|
|
const readSpy = vi.spyOn(dbStore, 'read');
|
|
|
|
const execution = await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
maxDataSizeBytes: ONE_MB,
|
|
});
|
|
|
|
expect(execution?.dataTooLargeToDisplay).toBe(true);
|
|
expect(execution?.data?.resultData?.runData).toEqual({});
|
|
// the DB size check rejects it without loading the run data
|
|
expect(readSpy).not.toHaveBeenCalled();
|
|
|
|
readSpy.mockRestore();
|
|
});
|
|
|
|
it('skips reading the run data and loads only the workflow snapshot when oversized', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const dbStore = Container.get(DbStore);
|
|
const { executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const readSpy = vi.spyOn(dbStore, 'read');
|
|
const readWorkflowDataSpy = vi.spyOn(dbStore, 'readWorkflowData');
|
|
|
|
await executionPersistence.findSingleExecution(executionId, {
|
|
includeData: true,
|
|
unflattenData: true,
|
|
maxDataSizeBytes: ONE_MB,
|
|
});
|
|
|
|
// the run data is never read; only the (small) snapshot is
|
|
expect(readSpy).not.toHaveBeenCalled();
|
|
expect(readWorkflowDataSpy).toHaveBeenCalledTimes(1);
|
|
|
|
readSpy.mockRestore();
|
|
readWorkflowDataSpy.mockRestore();
|
|
});
|
|
|
|
it('does not guard operational reads (findWithUnflattenedData loads full data)', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { workflow, executionId } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const execution = await executionPersistence.findWithUnflattenedData(executionId, [
|
|
workflow.id,
|
|
]);
|
|
|
|
expect(execution?.dataTooLargeToDisplay).toBeFalsy();
|
|
expect(execution?.data?.resultData?.runData).toHaveProperty('bigNode');
|
|
});
|
|
|
|
it('findManyInWorkflows omits oversized data and flags it', async () => {
|
|
const executionPersistence = Container.get(ExecutionPersistence);
|
|
const { workflow } = await createSizedExecution(2 * ONE_MB);
|
|
|
|
const executions = (await executionPersistence.findManyInWorkflows(
|
|
[workflow.id],
|
|
{ limit: 10, includeData: true },
|
|
ONE_MB,
|
|
)) as IExecutionResponse[];
|
|
|
|
expect(executions[0]?.dataTooLargeToDisplay).toBe(true);
|
|
expect(executions[0]?.data?.resultData?.runData).toEqual({});
|
|
});
|
|
});
|
|
});
|