n8n/packages/core/test/NodeExecuteFunctions.test.ts
Iván Ovejero cd08c8e4c6
refactor(core): Implement soft-deletions for executions (#7092)
Based on #7065 | Story: https://linear.app/n8n/issue/PAY-771

n8n on filesystem mode marks binary data to delete on manual execution
deletion, on unsaved execution completion, and on every execution
pruning cycle. We later prune binary data in a separate cycle via these
marker files, based on the configured TTL. In the context of introducing
an S3 client to manage binary data, the filesystem mode's mark-and-prune
setup is too tightly coupled to the general binary data management
client interface.

This PR...
- Ensures the deletion of an execution causes the deletion of any binary
data associated to it. This does away with the need for binary data TTL
and simplifies the filesystem mode's mark-and-prune setup.
- Refactors all execution deletions (including pruning) to cause soft
deletions, hard-deletes soft-deleted executions based on the existing
pruning config, and adjusts execution endpoints to filter out
soft-deleted executions. This reduces DB load, and keeps binary data
around long enough for users to access it when building workflows with
unsaved executions.
- Moves all execution pruning work from an execution lifecycle hook to
`execution.repository.ts`. This keeps related logic in a single place.
- Removes all marking logic from the binary data manager. This
simplifies the interface that the S3 client will meet.
- Adds basic sanity-check tests to pruning logic and execution deletion.

Out of scope:

- Improving existing pruning logic.
- Improving existing execution repository logic.
- Adjusting dir structure for filesystem mode.

---------

Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <aditya@netroy.in>
2023-09-20 15:21:42 +02:00

284 lines
8.6 KiB
TypeScript

import { BinaryDataManager } from '@/BinaryDataManager';
import {
getBinaryDataBuffer,
parseIncomingMessage,
proxyRequestToAxios,
setBinaryDataBuffer,
} from '@/NodeExecuteFunctions';
import { mkdtempSync, readFileSync } from 'fs';
import type { IncomingMessage } from 'http';
import { mock } from 'jest-mock-extended';
import type {
IBinaryData,
INode,
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
Workflow,
WorkflowHooks,
} from 'n8n-workflow';
import nock from 'nock';
import { tmpdir } from 'os';
import { join } from 'path';
import { initLogger } from './helpers/utils';
const temporaryDir = mkdtempSync(join(tmpdir(), 'n8n'));
describe('NodeExecuteFunctions', () => {
describe('test binary data helper methods', () => {
// Reset BinaryDataManager for each run. This is a dirty operation, as individual managers are not cleaned.
beforeEach(() => {
BinaryDataManager.instance = undefined;
});
test("test getBinaryDataBuffer(...) & setBinaryDataBuffer(...) methods in 'default' mode", async () => {
// Setup a 'default' binary data manager instance
await BinaryDataManager.init({
mode: 'default',
availableModes: 'default',
localStoragePath: temporaryDir,
});
// Set our binary data buffer
const inputData: Buffer = Buffer.from('This is some binary data', 'utf8');
const setBinaryDataBufferResponse: IBinaryData = await setBinaryDataBuffer(
{
mimeType: 'txt',
data: 'This should be overwritten by the actual payload in the response',
},
inputData,
'executionId',
);
// Expect our return object to contain the base64 encoding of the input data, as it should be stored in memory.
expect(setBinaryDataBufferResponse.data).toEqual(inputData.toString('base64'));
// Now, re-fetch our data.
// An ITaskDataConnections object is used to share data between nodes. The top level property, 'main', represents the successful output object from a previous node.
const taskDataConnectionsInput: ITaskDataConnections = {
main: [],
};
// We add an input set, with one item at index 0, to this input. It contains an empty json payload and our binary data.
taskDataConnectionsInput.main.push([
{
json: {},
binary: {
data: setBinaryDataBufferResponse,
},
},
]);
// Now, lets fetch our data! The item will be item index 0.
const getBinaryDataBufferResponse: Buffer = await getBinaryDataBuffer(
taskDataConnectionsInput,
0,
'data',
0,
);
expect(getBinaryDataBufferResponse).toEqual(inputData);
});
test("test getBinaryDataBuffer(...) & setBinaryDataBuffer(...) methods in 'filesystem' mode", async () => {
// Setup a 'filesystem' binary data manager instance
await BinaryDataManager.init({
mode: 'filesystem',
availableModes: 'filesystem',
localStoragePath: temporaryDir,
});
// Set our binary data buffer
const inputData: Buffer = Buffer.from('This is some binary data', 'utf8');
const setBinaryDataBufferResponse: IBinaryData = await setBinaryDataBuffer(
{
mimeType: 'txt',
data: 'This should be overwritten with the name of the configured data manager',
},
inputData,
'executionId',
);
// Expect our return object to contain the name of the configured data manager.
expect(setBinaryDataBufferResponse.data).toEqual('filesystem');
// Ensure that the input data was successfully persisted to disk.
expect(
readFileSync(
`${temporaryDir}/${setBinaryDataBufferResponse.id?.replace('filesystem:', '')}`,
),
).toEqual(inputData);
// Now, re-fetch our data.
// An ITaskDataConnections object is used to share data between nodes. The top level property, 'main', represents the successful output object from a previous node.
const taskDataConnectionsInput: ITaskDataConnections = {
main: [],
};
// We add an input set, with one item at index 0, to this input. It contains an empty json payload and our binary data.
taskDataConnectionsInput.main.push([
{
json: {},
binary: {
data: setBinaryDataBufferResponse,
},
},
]);
// Now, lets fetch our data! The item will be item index 0.
const getBinaryDataBufferResponse: Buffer = await getBinaryDataBuffer(
taskDataConnectionsInput,
0,
'data',
0,
);
expect(getBinaryDataBufferResponse).toEqual(inputData);
});
});
describe('parseIncomingMessage', () => {
it('parses valid content-type header', () => {
const message = mock<IncomingMessage>({
headers: { 'content-type': 'application/json', 'content-disposition': undefined },
});
parseIncomingMessage(message);
expect(message.contentType).toEqual('application/json');
});
it('parses valid content-type header with parameters', () => {
const message = mock<IncomingMessage>({
headers: {
'content-type': 'application/json; charset=utf-8',
'content-disposition': undefined,
},
});
parseIncomingMessage(message);
expect(message.contentType).toEqual('application/json');
});
it('parses valid content-disposition header with filename*', () => {
const message = mock<IncomingMessage>({
headers: {
'content-type': undefined,
'content-disposition':
'attachment; filename="screenshot%20(1).png"; filename*=UTF-8\'\'screenshot%20(1).png',
},
});
parseIncomingMessage(message);
expect(message.contentDisposition).toEqual({
filename: 'screenshot (1).png',
type: 'attachment',
});
});
it('parses valid content-disposition header with filename and trailing ";"', () => {
const message = mock<IncomingMessage>({
headers: {
'content-type': undefined,
'content-disposition': 'inline; filename="screenshot%20(1).png";',
},
});
parseIncomingMessage(message);
expect(message.contentDisposition).toEqual({
filename: 'screenshot (1).png',
type: 'inline',
});
});
it('parses non standard content-disposition with missing type', () => {
const message = mock<IncomingMessage>({
headers: {
'content-type': undefined,
'content-disposition': 'filename="screenshot%20(1).png";',
},
});
parseIncomingMessage(message);
expect(message.contentDisposition).toEqual({
filename: 'screenshot (1).png',
type: 'attachment',
});
});
});
describe('proxyRequestToAxios', () => {
const baseUrl = 'http://example.de';
const workflow = mock<Workflow>();
const hooks = mock<WorkflowHooks>();
const additionalData = mock<IWorkflowExecuteAdditionalData>({ hooks });
const node = mock<INode>();
beforeEach(() => {
initLogger();
hooks.executeHookFunctions.mockClear();
});
test('should not throw if the response status is 200', async () => {
nock(baseUrl).get('/test').reply(200);
await proxyRequestToAxios(workflow, additionalData, node, `${baseUrl}/test`);
expect(hooks.executeHookFunctions).toHaveBeenCalledWith('nodeFetchedData', [
workflow.id,
node,
]);
});
test('should throw if the response status is 403', async () => {
const headers = { 'content-type': 'text/plain' };
nock(baseUrl).get('/test').reply(403, 'Forbidden', headers);
try {
await proxyRequestToAxios(workflow, additionalData, node, `${baseUrl}/test`);
} catch (error) {
expect(error.statusCode).toEqual(403);
expect(error.request).toBeUndefined();
expect(error.response).toMatchObject({ headers, status: 403 });
expect(error.options).toMatchObject({
headers: { Accept: '*/*' },
method: 'get',
url: 'http://example.de/test',
});
expect(error.config).toBeUndefined();
expect(error.message).toEqual('403 - "Forbidden"');
}
expect(hooks.executeHookFunctions).not.toHaveBeenCalled();
});
test('should not throw if the response status is 404, but `simple` option is set to `false`', async () => {
nock(baseUrl).get('/test').reply(404, 'Not Found');
const response = await proxyRequestToAxios(workflow, additionalData, node, {
url: `${baseUrl}/test`,
simple: false,
});
expect(response).toEqual('Not Found');
expect(hooks.executeHookFunctions).toHaveBeenCalledWith('nodeFetchedData', [
workflow.id,
node,
]);
});
test('should return full response when `resolveWithFullResponse` is set to true', async () => {
nock(baseUrl).get('/test').reply(404, 'Not Found');
const response = await proxyRequestToAxios(workflow, additionalData, node, {
url: `${baseUrl}/test`,
resolveWithFullResponse: true,
simple: false,
});
expect(response).toMatchObject({
body: 'Not Found',
headers: {},
statusCode: 404,
statusMessage: null,
});
expect(hooks.executeHookFunctions).toHaveBeenCalledWith('nodeFetchedData', [
workflow.id,
node,
]);
});
});
});