mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-01-11 21:07:28 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:n8n-io/n8n into ado-2808-1

Browse source
This commit is contained in:
Mutasem Aldmour 2024-11-07 11:15:40 +01:00
parent f456191c73 3111dece72
commit 67e09e2988
No known key found for this signature in database
GPG key ID: 3DFA8122BB7FD6B8
41 changed files with 1561 additions and 1045 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.github/workflows/release-publish.yml vendored
View file

@ -126,7 +126,7 @@ jobs:
body: ${{github.event.pull_request.body}}
create-sentry-release:
name: Create release on Sentry
name: Create a Sentry Release
needs: [publish-to-npm, publish-to-docker-hub]
runs-on: ubuntu-latest
if: github.event.pull_request.merged == true
@ -136,6 +136,7 @@ jobs:
SENTRY_ORG: ${{ secrets.SENTRY_ORG }}
steps:
- uses: actions/checkout@v4.1.1
- name: Restore cached build artifacts
uses: actions/cache/restore@v4.0.0
with:
@ -147,7 +148,7 @@ jobs:
continue-on-error: true
with:
projects: ${{ secrets.SENTRY_FRONTEND_PROJECT }}
version: {{ needs.publish-to-npm.outputs.release }}
version: ${{ needs.publish-to-npm.outputs.release }}
sourcemaps: packages/editor-ui/dist
- name: Create a backend release
@ -155,7 +156,7 @@ jobs:
continue-on-error: true
with:
projects: ${{ secrets.SENTRY_BACKEND_PROJECT }}
version: {{ needs.publish-to-npm.outputs.release }}
version: ${{ needs.publish-to-npm.outputs.release }}
sourcemaps: packages/cli/dist packages/core/dist packages/nodes-base/dist packages/@n8n/n8n-nodes-langchain/dist
trigger-release-note:

36
CHANGELOG.md
View file

@ -1,3 +1,39 @@
# [1.67.0](https://github.com/n8n-io/n8n/compare/n8n@1.66.0...n8n@1.67.0) (2024-11-06)
### Bug Fixes
* Bring back nodes panel telemetry events ([#11456](https://github.com/n8n-io/n8n/issues/11456)) ([130c942](https://github.com/n8n-io/n8n/commit/130c942f633788d1b2f937d6fea342d4450c6e3d))
* **core:** Account for double quotes in instance base URL ([#11495](https://github.com/n8n-io/n8n/issues/11495)) ([c5191e6](https://github.com/n8n-io/n8n/commit/c5191e697a9a9ebfa2b67587cd01b5835ebf6ea8))
* **core:** Do not delete waiting executions when saving of successful executions is disabled ([#11458](https://github.com/n8n-io/n8n/issues/11458)) ([e8757e5](https://github.com/n8n-io/n8n/commit/e8757e58f69e091ac3d2a2f8e8c8e33ac57c1e47))
* **core:** Don't send a `executionFinished` event to the browser with no run data if the execution has already been cleaned up ([#11502](https://github.com/n8n-io/n8n/issues/11502)) ([d1153f5](https://github.com/n8n-io/n8n/commit/d1153f51e80911cbc8f34ba5f038f349b75295c3))
* **core:** Include `projectId` in range query middleware ([#11590](https://github.com/n8n-io/n8n/issues/11590)) ([a6070af](https://github.com/n8n-io/n8n/commit/a6070afdda29631fd36e5213f52bf815268bcda4))
* **core:** Save exeution progress for waiting executions, even when progress saving is disabled ([#11535](https://github.com/n8n-io/n8n/issues/11535)) ([6b9353c](https://github.com/n8n-io/n8n/commit/6b9353c80f61ab36945fff434d98242dc1cab7b3))
* **core:** Use the correct docs URL for regular nodes when used as tools ([#11529](https://github.com/n8n-io/n8n/issues/11529)) ([a092b8e](https://github.com/n8n-io/n8n/commit/a092b8e972e1253d92df416f19096a045858e7c1))
* **Edit Image Node:** Fix Text operation by setting Arial as default font ([#11125](https://github.com/n8n-io/n8n/issues/11125)) ([60c1ace](https://github.com/n8n-io/n8n/commit/60c1ace64be29d651ce7b777fbb576598e38b9d7))
* **editor:** Auto focus first fields on SignIn, SignUp and ForgotMyPassword views ([#11445](https://github.com/n8n-io/n8n/issues/11445)) ([5b5bd72](https://github.com/n8n-io/n8n/commit/5b5bd7291dde17880b7699f7e6832938599ffd8f))
* **editor:** Do not overwrite the webhookId in the new canvas ([#11562](https://github.com/n8n-io/n8n/issues/11562)) ([dfd785b](https://github.com/n8n-io/n8n/commit/dfd785bc0894257eb6e62b0dd8f71248c27aae53))
* **editor:** Ensure Enter key on Cancel button correctly cancels node rename ([#11563](https://github.com/n8n-io/n8n/issues/11563)) ([be05ae3](https://github.com/n8n-io/n8n/commit/be05ae36e7790156cb48b317fc254ae46a3b2d8c))
* **editor:** Fix emitting `n8nReady` notification via `postmessage` on new canvas ([#11558](https://github.com/n8n-io/n8n/issues/11558)) ([463d101](https://github.com/n8n-io/n8n/commit/463d101f3592e6df4afd66c4d0fde0cb4aec34cc))
* **editor:** Fix run index input for RunData view in sub-nodes ([#11538](https://github.com/n8n-io/n8n/issues/11538)) ([434d31c](https://github.com/n8n-io/n8n/commit/434d31ce928342d52b6ab8b78639afd7829216d4))
* **editor:** Fix selected credential being overwritten in NDV ([#11496](https://github.com/n8n-io/n8n/issues/11496)) ([a26c0e2](https://github.com/n8n-io/n8n/commit/a26c0e2c3c7da87bfaba9737a967aa0070810d85))
* **editor:** Keep workflow pristine after load on new canvas ([#11579](https://github.com/n8n-io/n8n/issues/11579)) ([7254359](https://github.com/n8n-io/n8n/commit/7254359855b89769613cd5cc24dbb4f45a7cc76f))
* Show Pinned data in demo mode ([#11490](https://github.com/n8n-io/n8n/issues/11490)) ([ca2a583](https://github.com/n8n-io/n8n/commit/ca2a583b5cbb0cba3ecb694261806de16547aa91))
* Toast not aligned to the bottom when AI assistant disable ([#11549](https://github.com/n8n-io/n8n/issues/11549)) ([e80f7e0](https://github.com/n8n-io/n8n/commit/e80f7e0a02a972379f73af6a44de11768081086e))
### Features
* Add Rapid7 InsightVm credentials ([#11462](https://github.com/n8n-io/n8n/issues/11462)) ([46eceab](https://github.com/n8n-io/n8n/commit/46eceabc27ac219b11b85c16c533a2cff848c5dd))
* **AI Transform Node:** UX improvements ([#11280](https://github.com/n8n-io/n8n/issues/11280)) ([8a48407](https://github.com/n8n-io/n8n/commit/8a484077af3d3e1fe2d1b90b1ea9edf4ba41fcb6))
* **Anthropic Chat Model Node:** Add support for Haiku 3.5 ([#11551](https://github.com/n8n-io/n8n/issues/11551)) ([8b39825](https://github.com/n8n-io/n8n/commit/8b398256a81594a52f20f8eb8adf8ff205209bc1))
* **Convert to File Node:** Add delimiter convert to csv ([#11556](https://github.com/n8n-io/n8n/issues/11556)) ([63d454b](https://github.com/n8n-io/n8n/commit/63d454b776c092ff8c6c521a7e083774adb8f649))
* **editor:** Update panning and selection keybindings on new canvas ([#11534](https://github.com/n8n-io/n8n/issues/11534)) ([5e2e205](https://github.com/n8n-io/n8n/commit/5e2e205394adf76faf02aee2d4f21df71848e1d4))
* **Gmail Trigger Node:** Add filter option to include drafts ([#11441](https://github.com/n8n-io/n8n/issues/11441)) ([7a2be77](https://github.com/n8n-io/n8n/commit/7a2be77f384a32ede3acad8fe24fb89227c058bf))
* **Intercom Node:** Update credential to new style ([#11485](https://github.com/n8n-io/n8n/issues/11485)) ([b137e13](https://github.com/n8n-io/n8n/commit/b137e13845f0714ebf7421c837f5ab104b66709b))
# [1.66.0](https://github.com/n8n-io/n8n/compare/n8n@1.65.0...n8n@1.66.0) (2024-10-31)

2
package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-monorepo",
"version": "1.66.0",
"version": "1.67.0",
"private": true,
"engines": {
"node": ">=20.15",

2
packages/@n8n/chat/package.json
View file

@ -1,6 +1,6 @@
{
"name": "@n8n/chat",
"version": "0.29.0",
"version": "0.30.0",
"scripts": {
"dev": "pnpm run storybook",
"build": "pnpm build:vite && pnpm build:bundle",

2
packages/@n8n/config/package.json
View file

@ -1,6 +1,6 @@
{
"name": "@n8n/config",
"version": "1.16.0",
"version": "1.17.0",
"scripts": {
"clean": "rimraf dist .turbo",
"dev": "pnpm watch",

4
packages/@n8n/config/src/configs/runners.config.ts
View file

@ -50,4 +50,8 @@ export class TaskRunnersConfig {
/** How many concurrent tasks can a runner execute at a time */
@Env('N8N_RUNNERS_MAX_CONCURRENCY')
maxConcurrency: number = 5;
/** Should the output of deduplication be asserted for correctness */
@Env('N8N_RUNNERS_ASSERT_DEDUPLICATION_OUTPUT')
assertDeduplicationOutput: boolean = false;
}

1
packages/@n8n/config/test/config.test.ts
View file

@ -233,6 +233,7 @@ describe('GlobalConfig', () => {
launcherRunner: 'javascript',
maxOldSpaceSize: '',
maxConcurrency: 5,
assertDeduplicationOutput: false,
},
sentry: {
backendDsn: '',

26
packages/@n8n/nodes-langchain/nodes/agents/Agent/agents/ToolsAgent/execute.ts
View file

@ -206,10 +206,28 @@ export async function toolsAgentExecute(this: IExecuteFunctions): Promise<INodeE
// If the steps are an AgentFinish and the outputParser is defined it must mean that the LLM didn't use `format_final_response` tool so we will try to parse the output manually
if (outputParser && typeof steps === 'object' && (steps as AgentFinish).returnValues) {
const finalResponse = (steps as AgentFinish).returnValues;
const returnValues = (await outputParser.parse(finalResponse as unknown as string)) as Record<
string,
unknown
>;
let parserInput: string;
if (finalResponse instanceof Object) {
if ('output' in finalResponse) {
try {
// If the output is an object, we will try to parse it as JSON
// this is because parser expects stringified JSON object like { "output": { .... } }
// so we try to parse the output before wrapping it and then stringify it
parserInput = JSON.stringify({ output: jsonParse(finalResponse.output) });
} catch (error) {
// If parsing of the output fails, we will use the raw output
parserInput = finalResponse.output;
}
} else {
// If the output is not an object, we will stringify it as it is
parserInput = JSON.stringify(finalResponse);
}
} else {
parserInput = finalResponse;
}
const returnValues = (await outputParser.parse(parserInput)) as Record<string, unknown>;
return handleParsedStepOutput(returnValues);
}
return handleAgentFinishOutput(steps);

39
packages/@n8n/nodes-langchain/nodes/output_parser/OutputParserAutofixing/OutputParserAutofixing.node.ts
View file

@ -1,5 +1,6 @@
import type { BaseLanguageModel } from '@langchain/core/language_models/base';
import { NodeConnectionType } from 'n8n-workflow';
import { PromptTemplate } from '@langchain/core/prompts';
import { NodeConnectionType, NodeOperationError } from 'n8n-workflow';
import type {
ISupplyDataFunctions,
INodeType,
@ -7,6 +8,7 @@ import type {
SupplyData,
} from 'n8n-workflow';
import { NAIVE_FIX_PROMPT } from './prompt';
import {
N8nOutputFixingParser,
type N8nStructuredOutputParser,
@ -65,6 +67,27 @@ export class OutputParserAutofixing implements INodeType {
default: '',
},
getConnectionHintNoticeField([NodeConnectionType.AiChain, NodeConnectionType.AiAgent]),
{
displayName: 'Options',
name: 'options',
type: 'collection',
placeholder: 'Add Option',
default: {},
options: [
{
displayName: 'Retry Prompt',
name: 'prompt',
type: 'string',
default: NAIVE_FIX_PROMPT,
typeOptions: {
rows: 10,
},
hint: 'Should include "{error}", "{instructions}", and "{completion}" placeholders',
description:
'Prompt template used for fixing the output. Uses placeholders: "{instructions}" for parsing rules, "{completion}" for the failed attempt, and "{error}" for the validation error message.',
},
],
},
],
};
@ -77,8 +100,20 @@ export class OutputParserAutofixing implements INodeType {
NodeConnectionType.AiOutputParser,
itemIndex,
)) as N8nStructuredOutputParser;
const prompt = this.getNodeParameter('options.prompt', itemIndex, NAIVE_FIX_PROMPT) as string;
const parser = new N8nOutputFixingParser(this, model, outputParser);
if (prompt.length === 0 || !prompt.includes('{error}')) {
throw new NodeOperationError(
this.getNode(),
'Auto-fixing parser prompt has to contain {error} placeholder',
);
}
const parser = new N8nOutputFixingParser(
this,
model,
outputParser,
PromptTemplate.fromTemplate(prompt),
);
return {
response: parser,

16
packages/@n8n/nodes-langchain/nodes/output_parser/OutputParserAutofixing/prompt.ts Normal file
View file

@ -0,0 +1,16 @@
export const NAIVE_FIX_PROMPT = `Instructions:
--------------
{instructions}
--------------
Completion:
--------------
{completion}
--------------
Above, the Completion did not satisfy the constraints given in the Instructions.
Error:
--------------
{error}
--------------
Please try again. Please only respond with an answer that satisfies the constraints laid out in the Instructions:`;

168
packages/@n8n/nodes-langchain/nodes/output_parser/OutputParserAutofixing/test/OutputParserAutofixing.node.test.ts
View file

@ -1,15 +1,19 @@
/* eslint-disable @typescript-eslint/unbound-method */
/* eslint-disable @typescript-eslint/no-unsafe-call */
import type { BaseLanguageModel } from '@langchain/core/language_models/base';
import { OutputParserException } from '@langchain/core/output_parsers';
import type { MockProxy } from 'jest-mock-extended';
import { mock } from 'jest-mock-extended';
import { normalizeItems } from 'n8n-core';
import type { IExecuteFunctions, IWorkflowDataProxyData } from 'n8n-workflow';
import { ApplicationError, NodeConnectionType } from 'n8n-workflow';
import { ApplicationError, NodeConnectionType, NodeOperationError } from 'n8n-workflow';
import { N8nOutputFixingParser } from '../../../../utils/output_parsers/N8nOutputParser';
import type { N8nStructuredOutputParser } from '../../../../utils/output_parsers/N8nOutputParser';
import type {
N8nOutputFixingParser,
N8nStructuredOutputParser,
} from '../../../../utils/output_parsers/N8nOutputParser';
import { OutputParserAutofixing } from '../OutputParserAutofixing.node';
import { NAIVE_FIX_PROMPT } from '../prompt';
describe('OutputParserAutofixing', () => {
let outputParser: OutputParserAutofixing;
@ -34,6 +38,13 @@ describe('OutputParserAutofixing', () => {
throw new ApplicationError('Unexpected connection type');
});
thisArg.getNodeParameter.mockReset();
thisArg.getNodeParameter.mockImplementation((parameterName) => {
if (parameterName === 'options.prompt') {
return NAIVE_FIX_PROMPT;
}
throw new ApplicationError('Not implemented');
});
});
afterEach(() => {
@ -48,73 +59,132 @@ describe('OutputParserAutofixing', () => {
});
}
it('should successfully parse valid output without needing to fix it', async () => {
const validOutput = { name: 'Alice', age: 25 };
describe('Configuration', () => {
it('should throw error when prompt template does not contain {error} placeholder', async () => {
thisArg.getNodeParameter.mockImplementation((parameterName) => {
if (parameterName === 'options.prompt') {
return 'Invalid prompt without error placeholder';
}
throw new ApplicationError('Not implemented');
});
mockStructuredOutputParser.parse.mockResolvedValueOnce(validOutput);
await expect(outputParser.supplyData.call(thisArg, 0)).rejects.toThrow(
new NodeOperationError(
thisArg.getNode(),
'Auto-fixing parser prompt has to contain {error} placeholder',
),
);
});
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
it('should throw error when prompt template is empty', async () => {
thisArg.getNodeParameter.mockImplementation((parameterName) => {
if (parameterName === 'options.prompt') {
return '';
}
throw new ApplicationError('Not implemented');
});
// Ensure the response contains the output-fixing parser
expect(response).toBeDefined();
expect(response).toBeInstanceOf(N8nOutputFixingParser);
await expect(outputParser.supplyData.call(thisArg, 0)).rejects.toThrow(
new NodeOperationError(
thisArg.getNode(),
'Auto-fixing parser prompt has to contain {error} placeholder',
),
);
});
const result = await response.parse('{"name": "Alice", "age": 25}');
it('should use default prompt when none specified', async () => {
thisArg.getNodeParameter.mockImplementation((parameterName) => {
if (parameterName === 'options.prompt') {
return NAIVE_FIX_PROMPT;
}
throw new ApplicationError('Not implemented');
});
// Validate that the parser succeeds without retry
expect(result).toEqual(validOutput);
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(1); // Only one call to parse
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
expect(response).toBeDefined();
});
});
it('should throw an error when both structured parser and fixing parser fail', async () => {
mockStructuredOutputParser.parse
.mockRejectedValueOnce(new Error('Invalid JSON')) // First attempt fails
.mockRejectedValueOnce(new Error('Fixing attempt failed')); // Second attempt fails
describe('Parsing', () => {
it('should successfully parse valid output without needing to fix it', async () => {
const validOutput = { name: 'Alice', age: 25 };
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
mockStructuredOutputParser.parse.mockResolvedValueOnce(validOutput);
response.getRetryChain = getMockedRetryChain('{}');
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
await expect(response.parse('Invalid JSON string')).rejects.toThrow('Fixing attempt failed');
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(2);
});
const result = await response.parse('{"name": "Alice", "age": 25}');
it('should reject on the first attempt and succeed on retry with the parsed content', async () => {
const validOutput = { name: 'Bob', age: 28 };
expect(result).toEqual(validOutput);
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(1);
});
mockStructuredOutputParser.parse.mockRejectedValueOnce(new Error('Invalid JSON'));
it('should not retry on non-OutputParserException errors', async () => {
const error = new Error('Some other error');
mockStructuredOutputParser.parse.mockRejectedValueOnce(error);
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
response.getRetryChain = getMockedRetryChain(JSON.stringify(validOutput));
await expect(response.parse('Invalid JSON string')).rejects.toThrow(error);
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(1);
});
mockStructuredOutputParser.parse.mockResolvedValueOnce(validOutput);
it('should retry on OutputParserException and succeed', async () => {
const validOutput = { name: 'Bob', age: 28 };
const result = await response.parse('Invalid JSON string');
mockStructuredOutputParser.parse
.mockRejectedValueOnce(new OutputParserException('Invalid JSON'))
.mockResolvedValueOnce(validOutput);
expect(result).toEqual(validOutput);
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(2); // First fails, second succeeds
});
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
it('should handle non-JSON formatted response from fixing parser', async () => {
mockStructuredOutputParser.parse.mockRejectedValueOnce(new Error('Invalid JSON'));
response.getRetryChain = getMockedRetryChain(JSON.stringify(validOutput));
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
const result = await response.parse('Invalid JSON string');
response.getRetryChain = getMockedRetryChain('This is not JSON');
expect(result).toEqual(validOutput);
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(2);
});
mockStructuredOutputParser.parse.mockRejectedValueOnce(new Error('Unexpected token'));
it('should handle failed retry attempt', async () => {
mockStructuredOutputParser.parse
.mockRejectedValueOnce(new OutputParserException('Invalid JSON'))
.mockRejectedValueOnce(new Error('Still invalid JSON'));
// Expect the structured parser to throw an error on invalid JSON from retry
await expect(response.parse('Invalid JSON string')).rejects.toThrow('Unexpected token');
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(2); // First fails, second tries and fails
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
response.getRetryChain = getMockedRetryChain('Still not valid JSON');
await expect(response.parse('Invalid JSON string')).rejects.toThrow('Still invalid JSON');
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(2);
});
it('should throw non-OutputParserException errors immediately without retry', async () => {
const customError = new Error('Database connection error');
const retryChainSpy = jest.fn();
mockStructuredOutputParser.parse.mockRejectedValueOnce(customError);
const { response } = (await outputParser.supplyData.call(thisArg, 0)) as {
response: N8nOutputFixingParser;
};
response.getRetryChain = retryChainSpy;
await expect(response.parse('Some input')).rejects.toThrow('Database connection error');
expect(mockStructuredOutputParser.parse).toHaveBeenCalledTimes(1);
expect(retryChainSpy).not.toHaveBeenCalled();
});
});
});

2
packages/@n8n/nodes-langchain/package.json
View file

@ -1,6 +1,6 @@
{
"name": "@n8n/n8n-nodes-langchain",
"version": "1.66.0",
"version": "1.67.0",
"description": "",
"main": "index.js",
"scripts": {

12
packages/@n8n/nodes-langchain/utils/output_parsers/N8nOutputFixingParser.ts
View file

@ -1,12 +1,12 @@
import type { Callbacks } from '@langchain/core/callbacks/manager';
import type { BaseLanguageModel } from '@langchain/core/language_models/base';
import type { AIMessage } from '@langchain/core/messages';
import { BaseOutputParser } from '@langchain/core/output_parsers';
import { BaseOutputParser, OutputParserException } from '@langchain/core/output_parsers';
import type { PromptTemplate } from '@langchain/core/prompts';
import type { ISupplyDataFunctions } from 'n8n-workflow';
import { NodeConnectionType } from 'n8n-workflow';
import type { N8nStructuredOutputParser } from './N8nStructuredOutputParser';
import { NAIVE_FIX_PROMPT } from './prompt';
import { logAiEvent } from '../helpers';
export class N8nOutputFixingParser extends BaseOutputParser {
@ -16,12 +16,13 @@ export class N8nOutputFixingParser extends BaseOutputParser {
private context: ISupplyDataFunctions,
private model: BaseLanguageModel,
private outputParser: N8nStructuredOutputParser,
private fixPromptTemplate: PromptTemplate,
) {
super();
}
getRetryChain() {
return NAIVE_FIX_PROMPT.pipe(this.model);
return this.fixPromptTemplate.pipe(this.model);
}
/**
@ -47,11 +48,14 @@ export class N8nOutputFixingParser extends BaseOutputParser {
return response;
} catch (error) {
if (!(error instanceof OutputParserException)) {
throw error;
}
try {
// Second attempt: use retry chain to fix the output
const result = (await this.getRetryChain().invoke({
completion,
error,
error: error.message,
instructions: this.getFormatInstructions(),
})) as AIMessage;

2
packages/@n8n/task-runner/package.json
View file

@ -1,6 +1,6 @@
{
"name": "@n8n/task-runner",
"version": "1.4.0",
"version": "1.5.0",
"scripts": {
"clean": "rimraf dist .turbo",
"start": "node dist/start.js",

29
packages/@n8n/task-runner/src/data-request/data-request-response-reconstruct.ts Normal file
View file

@ -0,0 +1,29 @@
import type { IExecuteData, INodeExecutionData } from 'n8n-workflow';
import type { DataRequestResponse } from '@/runner-types';
/**
* Reconstructs data from a DataRequestResponse to the initial
* data structures.
*/
export class DataRequestResponseReconstruct {
/**
* Reconstructs `connectionInputData` from a DataRequestResponse
*/
reconstructConnectionInputData(
inputData: DataRequestResponse['inputData'],
): INodeExecutionData[] {
return inputData?.main?.[0] ?? [];
}
/**
* Reconstruct `executeData` from a DataRequestResponse
*/
reconstructExecuteData(response: DataRequestResponse): IExecuteData {
return {
data: response.inputData,
node: response.node,
source: response.connectionInputSource,
};
}
}

1
packages/@n8n/task-runner/src/index.ts
View file

@ -1,3 +1,4 @@
export * from './task-runner';
export * from './runner-types';
export * from './message-types';
export * from './data-request/data-request-response-reconstruct';

38
packages/@n8n/task-runner/src/js-task-runner/__tests__/js-task-runner.test.ts
View file

@ -3,15 +3,21 @@ import type { CodeExecutionMode, IDataObject } from 'n8n-workflow';
import fs from 'node:fs';
import { builtinModules } from 'node:module';
import type { JsRunnerConfig } from '@/config/js-runner-config';
import { MainConfig } from '@/config/main-config';
import { ExecutionError } from '@/js-task-runner/errors/execution-error';
import { ValidationError } from '@/js-task-runner/errors/validation-error';
import type { DataRequestResponse, JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { JSExecSettings } from '@/js-task-runner/js-task-runner';
import { JsTaskRunner } from '@/js-task-runner/js-task-runner';
import type { DataRequestResponse } from '@/runner-types';
import type { Task } from '@/task-runner';
import { newCodeTaskData, newTaskWithSettings, withPairedItem, wrapIntoJson } from './test-data';
import type { JsRunnerConfig } from '../../config/js-runner-config';
import { MainConfig } from '../../config/main-config';
import { ExecutionError } from '../errors/execution-error';
import {
newDataRequestResponse,
newTaskWithSettings,
withPairedItem,
wrapIntoJson,
} from './test-data';
jest.mock('ws');
@ -68,7 +74,7 @@ describe('JsTaskRunner', () => {
nodeMode: 'runOnceForAllItems',
...settings,
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson)),
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson)),
runner,
});
};
@ -91,7 +97,7 @@ describe('JsTaskRunner', () => {
nodeMode: 'runOnceForEachItem',
...settings,
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson)),
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson)),
runner,
});
};
@ -108,7 +114,7 @@ describe('JsTaskRunner', () => {
await execTaskWithParams({
task,
taskData: newCodeTaskData([wrapIntoJson({})]),
taskData: newDataRequestResponse([wrapIntoJson({})]),
});
expect(defaultTaskRunner.makeRpcCall).toHaveBeenCalledWith(task.taskId, 'logNodeOutput', [
@ -243,7 +249,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.VAR1 }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: false,
isProcessAvailable: true,
@ -262,7 +268,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.VAR1 }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: true,
isProcessAvailable: true,
@ -279,7 +285,7 @@ describe('JsTaskRunner', () => {
code: 'return Object.values($env).concat(Object.keys($env))',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: false,
isProcessAvailable: true,
@ -298,7 +304,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.N8N_RUNNERS_N8N_URI }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -313,7 +319,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: Buffer.from("test-buffer").toString() }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -325,7 +331,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: Buffer.from("test-buffer").toString() }',
nodeMode: 'runOnceForEachItem',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -771,7 +777,7 @@ describe('JsTaskRunner', () => {
code: 'unknown',
nodeMode,
}),
taskData: newCodeTaskData([wrapIntoJson({ a: 1 })]),
taskData: newDataRequestResponse([wrapIntoJson({ a: 1 })]),
}),
).rejects.toThrow(ExecutionError);
},
@ -793,7 +799,7 @@ describe('JsTaskRunner', () => {
jest.spyOn(runner, 'sendOffers').mockImplementation(() => {});
jest
.spyOn(runner, 'requestData')
.mockResolvedValue(newCodeTaskData([wrapIntoJson({ a: 1 })]));
.mockResolvedValue(newDataRequestResponse([wrapIntoJson({ a: 1 })]));
await runner.receivedSettings(taskId, task.settings);

29
packages/@n8n/task-runner/src/js-task-runner/__tests__/test-data.ts
View file

@ -2,7 +2,8 @@ import type { IDataObject, INode, INodeExecutionData, ITaskData } from 'n8n-work
import { NodeConnectionType } from 'n8n-workflow';
import { nanoid } from 'nanoid';
import type { DataRequestResponse, JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { DataRequestResponse } from '@/runner-types';
import type { Task } from '@/task-runner';
/**
@ -46,10 +47,10 @@ export const newTaskData = (opts: Partial<ITaskData> & Pick<ITaskData, 'source'>
});
/**
* Creates a new all code task data with the given options
* Creates a new data request response with the given options
*/
export const newCodeTaskData = (
codeNodeInputData: INodeExecutionData[],
export const newDataRequestResponse = (
inputData: INodeExecutionData[],
opts: Partial<DataRequestResponse> = {},
): DataRequestResponse => {
const codeNode = newNode({
@ -83,9 +84,8 @@ export const newCodeTaskData = (
nodes: [manualTriggerNode, codeNode],
},
inputData: {
main: [codeNodeInputData],
main: [inputData],
},
connectionInputData: codeNodeInputData,
node: codeNode,
runExecutionData: {
startData: {},
@ -95,7 +95,7 @@ export const newCodeTaskData = (
newTaskData({
source: [],
data: {
main: [codeNodeInputData],
main: [inputData],
},
}),
],
@ -137,14 +137,13 @@ export const newCodeTaskData = (
var: 'value',
},
},
executeData: {
node: codeNode,
data: {
main: [codeNodeInputData],
},
source: {
main: [{ previousNode: manualTriggerNode.name }],
},
connectionInputSource: {
main: [
{
previousNode: 'Trigger',
previousNodeOutput: 0,
},
],
},
...opts,
};

20
packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/__tests__/built-ins-parser.test.ts
View file

@ -1,8 +1,13 @@
import { getAdditionalKeys } from 'n8n-core';
import type { IDataObject, INodeType, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import type {
IDataObject,
IExecuteData,
INodeType,
IWorkflowExecuteAdditionalData,
} from 'n8n-workflow';
import { Workflow, WorkflowDataProxy } from 'n8n-workflow';
import { newCodeTaskData } from '../../__tests__/test-data';
import { newDataRequestResponse } from '../../__tests__/test-data';
import { BuiltInsParser } from '../built-ins-parser';
import { BuiltInsParserState } from '../built-ins-parser-state';
@ -159,7 +164,12 @@ describe('BuiltInsParser', () => {
describe('WorkflowDataProxy built-ins', () => {
it('should have a known list of built-ins', () => {
const data = newCodeTaskData([]);
const data = newDataRequestResponse([]);
const executeData: IExecuteData = {
data: {},
node: data.node,
source: data.connectionInputSource,
};
const dataProxy = new WorkflowDataProxy(
new Workflow({
...data.workflow,
@ -179,7 +189,7 @@ describe('BuiltInsParser', () => {
data.runIndex,
0,
data.activeNodeName,
data.connectionInputData,
[],
data.siblingParameters,
data.mode,
getAdditionalKeys(
@ -187,7 +197,7 @@ describe('BuiltInsParser', () => {
data.mode,
data.runExecutionData,
),
data.executeData,
executeData,
data.defaultReturnRunIndex,
data.selfData,
data.contextNodeName,

114
packages/@n8n/task-runner/src/js-task-runner/js-task-runner.ts
View file

@ -1,28 +1,25 @@
import { getAdditionalKeys } from 'n8n-core';
import {
WorkflowDataProxy,
// type IWorkflowDataProxyAdditionalKeys,
Workflow,
} from 'n8n-workflow';
import { WorkflowDataProxy, Workflow } from 'n8n-workflow';
import type {
CodeExecutionMode,
INode,
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
WorkflowParameters,
IDataObject,
IExecuteData,
INodeExecutionData,
INodeParameters,
IRunExecutionData,
WorkflowExecuteMode,
WorkflowParameters,
ITaskDataConnections,
INode,
IRunExecutionData,
EnvProviderState,
IExecuteData,
INodeTypeDescription,
} from 'n8n-workflow';
import * as a from 'node:assert';
import { runInNewContext, type Context } from 'node:vm';
import type { TaskResultData } from '@/runner-types';
import type { MainConfig } from '@/config/main-config';
import type { DataRequestResponse, PartialAdditionalData, TaskResultData } from '@/runner-types';
import { type Task, TaskRunner } from '@/task-runner';
import { BuiltInsParser } from './built-ins-parser/built-ins-parser';
@ -33,7 +30,7 @@ import { makeSerializable } from './errors/serializable-error';
import type { RequireResolver } from './require-resolver';
import { createRequireResolver } from './require-resolver';
import { validateRunForAllItemsOutput, validateRunForEachItemOutput } from './result-validation';
import type { MainConfig } from '../config/main-config';
import { DataRequestResponseReconstruct } from '../data-request/data-request-response-reconstruct';
export interface JSExecSettings {
code: string;
@ -45,34 +42,19 @@ export interface JSExecSettings {
mode: WorkflowExecuteMode;
}
export interface PartialAdditionalData {
executionId?: string;
restartExecutionId?: string;
restApiUrl: string;
instanceBaseUrl: string;
formWaitingBaseUrl: string;
webhookBaseUrl: string;
webhookWaitingBaseUrl: string;
webhookTestBaseUrl: string;
currentNodeParameters?: INodeParameters;
executionTimeoutTimestamp?: number;
userId?: string;
variables: IDataObject;
}
export interface DataRequestResponse {
export interface JsTaskData {
workflow: Omit<WorkflowParameters, 'nodeTypes'>;
inputData: ITaskDataConnections;
connectionInputData: INodeExecutionData[];
node: INode;
runExecutionData: IRunExecutionData;
runIndex: number;
itemIndex: number;
activeNodeName: string;
connectionInputData: INodeExecutionData[];
siblingParameters: INodeParameters;
mode: WorkflowExecuteMode;
envProviderState?: EnvProviderState;
envProviderState: EnvProviderState;
executeData?: IExecuteData;
defaultReturnRunIndex: number;
selfData: IDataObject;
@ -89,6 +71,8 @@ export class JsTaskRunner extends TaskRunner {
private readonly builtInsParser = new BuiltInsParser();
private readonly taskDataReconstruct = new DataRequestResponseReconstruct();
constructor(config: MainConfig, name = 'JS Task Runner') {
super({
taskType: 'javascript',
@ -115,33 +99,14 @@ export class JsTaskRunner extends TaskRunner {
? neededBuiltInsResult.result
: BuiltInsParserState.newNeedsAllDataState();
const data = await this.requestData<DataRequestResponse>(
const dataResponse = await this.requestData<DataRequestResponse>(
task.taskId,
neededBuiltIns.toDataRequestParams(),
);
/**
* We request node types only when we know a task needs all nodes, because
* needing all nodes means that the task relies on paired item functionality,
* which is the same requirement for needing node types.
*/
if (neededBuiltIns.needsAllNodes) {
const uniqueNodeTypes = new Map(
data.workflow.nodes.map((node) => [
`${node.type}|${node.typeVersion}`,
{ name: node.type, version: node.typeVersion },
]),
);
const data = this.reconstructTaskData(dataResponse);
const unknownNodeTypes = this.nodeTypes.onlyUnknown([...uniqueNodeTypes.values()]);
const nodeTypes = await this.requestNodeTypes<INodeTypeDescription[]>(
task.taskId,
unknownNodeTypes,
);
this.nodeTypes.addNodeTypeDescriptions(nodeTypes);
}
await this.requestNodeTypeIfNeeded(neededBuiltIns, data.workflow, task.taskId);
const workflowParams = data.workflow;
const workflow = new Workflow({
@ -201,7 +166,7 @@ export class JsTaskRunner extends TaskRunner {
private async runForAllItems(
taskId: string,
settings: JSExecSettings,
data: DataRequestResponse,
data: JsTaskData,
workflow: Workflow,
customConsole: CustomConsole,
): Promise<INodeExecutionData[]> {
@ -248,7 +213,7 @@ export class JsTaskRunner extends TaskRunner {
private async runForEachItem(
taskId: string,
settings: JSExecSettings,
data: DataRequestResponse,
data: JsTaskData,
workflow: Workflow,
customConsole: CustomConsole,
): Promise<INodeExecutionData[]> {
@ -315,7 +280,7 @@ export class JsTaskRunner extends TaskRunner {
return returnData;
}
private createDataProxy(data: DataRequestResponse, workflow: Workflow, itemIndex: number) {
private createDataProxy(data: JsTaskData, workflow: Workflow, itemIndex: number) {
return new WorkflowDataProxy(
workflow,
data.runExecutionData,
@ -359,4 +324,43 @@ export class JsTaskRunner extends TaskRunner {
return new ExecutionError({ message: JSON.stringify(error) });
}
private reconstructTaskData(response: DataRequestResponse): JsTaskData {
return {
...response,
connectionInputData: this.taskDataReconstruct.reconstructConnectionInputData(
response.inputData,
),
executeData: this.taskDataReconstruct.reconstructExecuteData(response),
};
}
private async requestNodeTypeIfNeeded(
neededBuiltIns: BuiltInsParserState,
workflow: JsTaskData['workflow'],
taskId: string,
) {
/**
* We request node types only when we know a task needs all nodes, because
* needing all nodes means that the task relies on paired item functionality,
* which is the same requirement for needing node types.
*/
if (neededBuiltIns.needsAllNodes) {
const uniqueNodeTypes = new Map(
workflow.nodes.map((node) => [
`${node.type}|${node.typeVersion}`,
{ name: node.type, version: node.typeVersion },
]),
);
const unknownNodeTypes = this.nodeTypes.onlyUnknown([...uniqueNodeTypes.values()]);
const nodeTypes = await this.requestNodeTypes<INodeTypeDescription[]>(
taskId,
unknownNodeTypes,
);
this.nodeTypes.addNodeTypeDescriptions(nodeTypes);
}
}
}

4
packages/@n8n/task-runner/src/runner-types.ts
View file

@ -8,6 +8,7 @@ import type {
INodeParameters,
IRunExecutionData,
ITaskDataConnections,
ITaskDataConnectionsSource,
IWorkflowExecuteAdditionalData,
Workflow,
WorkflowExecuteMode,
@ -29,17 +30,16 @@ export interface TaskDataRequestParams {
export interface DataRequestResponse {
workflow: Omit<WorkflowParameters, 'nodeTypes'>;
inputData: ITaskDataConnections;
connectionInputSource: ITaskDataConnectionsSource | null;
node: INode;
runExecutionData: IRunExecutionData;
runIndex: number;
itemIndex: number;
activeNodeName: string;
connectionInputData: INodeExecutionData[];
siblingParameters: INodeParameters;
mode: WorkflowExecuteMode;
envProviderState: EnvProviderState;
executeData?: IExecuteData;
defaultReturnRunIndex: number;
selfData: IDataObject;
contextNodeName: string;

2
packages/cli/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n",
"version": "1.66.0",
"version": "1.67.0",
"description": "n8n Workflow Automation Tool",
"main": "dist/index",
"types": "dist/index.d.ts",

12
packages/cli/src/runners/runner-types.ts
View file

@ -5,18 +5,6 @@ import type WebSocket from 'ws';
import type { TaskRunner } from './task-broker.service';
import type { AuthlessRequest } from '../requests';
/**
* Specifies what data should be included for a task data request.
*/
export interface TaskDataRequestParams {
dataOfNodes: string[] | 'all';
prevNode: boolean;
/** Whether input data for the node should be included */
input: boolean;
/** Whether env provider's state should be included */
env: boolean;
}
export interface DisconnectAnalyzer {
determineDisconnectReason(runnerId: TaskRunner['id']): Promise<Error>;
}

338
packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts
View file

@ -1,42 +1,10 @@
import type { TaskData } from '@n8n/task-runner';
import type { PartialAdditionalData, TaskData } from '@n8n/task-runner';
import { mock } from 'jest-mock-extended';
import type { IExecuteFunctions, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import { type INode, type INodeExecutionData, type Workflow } from 'n8n-workflow';
import type { Workflow } from 'n8n-workflow';
import { DataRequestResponseBuilder } from '../data-request-response-builder';
const triggerNode: INode = mock<INode>({
name: 'Trigger',
});
const debugHelperNode: INode = mock<INode>({
name: 'DebugHelper',
});
const codeNode: INode = mock<INode>({
name: 'Code',
});
const workflow: TaskData['workflow'] = mock<Workflow>();
const debugHelperNodeOutItems: INodeExecutionData[] = [
{
json: {
uid: 'abb74fd4-bef2-4fae-9d53-ea24e9eb3032',
email: 'Dan.Schmidt31@yahoo.com',
firstname: 'Toni',
lastname: 'Schuster',
password: 'Q!D6C2',
},
pairedItem: {
item: 0,
},
},
];
const codeNodeInputItems: INodeExecutionData[] = debugHelperNodeOutItems;
const connectionInputData: TaskData['connectionInputData'] = codeNodeInputItems;
const envProviderState: TaskData['envProviderState'] = mock<TaskData['envProviderState']>({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
const additionalData = mock<IWorkflowExecuteAdditionalData>({
const additionalData = mock<PartialAdditionalData>({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
@ -50,275 +18,57 @@ const additionalData = mock<IWorkflowExecuteAdditionalData>({
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
const executeFunctions = mock<IExecuteFunctions>();
/**
* Drawn with https://asciiflow.com/#/
* Task data for an execution of the following WF:
* where ►► denotes the currently being executing node.
* ►►
* ┌───────────┐ ┌─────────────┐ ┌────────┐
* │ Trigger ├──►│ DebugHelper ├───►│ Code │
* └───────────┘ └─────────────┘ └────────┘
*/
const taskData: TaskData = {
executeFunctions,
workflow,
connectionInputData,
inputData: {
main: [codeNodeInputItems],
},
itemIndex: 0,
activeNodeName: codeNode.name,
contextNodeName: codeNode.name,
defaultReturnRunIndex: -1,
mode: 'manual',
envProviderState,
node: codeNode,
runExecutionData: {
startData: {
destinationNode: codeNode.name,
runNodeFilter: [triggerNode.name, debugHelperNode.name, codeNode.name],
},
resultData: {
runData: {
[triggerNode.name]: [
{
hints: [],
startTime: 1730313407328,
executionTime: 1,
source: [],
executionStatus: 'success',
data: {
main: [[]],
},
},
],
[debugHelperNode.name]: [
{
hints: [],
startTime: 1730313407330,
executionTime: 1,
source: [
{
previousNode: triggerNode.name,
},
],
executionStatus: 'success',
data: {
main: [debugHelperNodeOutItems],
},
},
],
},
pinData: {},
},
executionData: {
contextData: {},
nodeExecutionStack: [],
metadata: {},
waitingExecution: {
[codeNode.name]: {
'0': {
main: [codeNodeInputItems],
},
},
},
waitingExecutionSource: {
[codeNode.name]: {
'0': {
main: [
{
previousNode: debugHelperNode.name,
},
],
},
},
},
},
},
runIndex: 0,
selfData: {},
siblingParameters: {},
executeData: {
node: codeNode,
data: {
main: [codeNodeInputItems],
},
source: {
main: [
{
previousNode: debugHelperNode.name,
previousNodeOutput: 0,
},
],
},
},
const workflow: TaskData['workflow'] = mock<Workflow>({
id: '1',
name: 'Test Workflow',
active: true,
connectionsBySourceNode: {},
nodes: {},
pinData: {},
settings: {},
staticData: {},
});
const taskData = mock<TaskData>({
additionalData,
} as const;
workflow,
});
describe('DataRequestResponseBuilder', () => {
const allDataParam: DataRequestResponseBuilder['requestParams'] = {
dataOfNodes: 'all',
env: true,
input: true,
prevNode: true,
};
const builder = new DataRequestResponseBuilder();
const newRequestParam = (opts: Partial<DataRequestResponseBuilder['requestParams']>) => ({
...allDataParam,
...opts,
});
it('picks only specific properties for additional data', () => {
const result = builder.buildFromTaskData(taskData);
describe('all data', () => {
it('should build the runExecutionData as is when everything is requested', () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(taskData, allDataParam);
const { runExecutionData } = dataRequestResponseBuilder.build();
expect(runExecutionData).toStrictEqual(taskData.runExecutionData);
expect(result.additionalData).toStrictEqual({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
variables: additionalData.variables,
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
});
describe('envProviderState', () => {
it("should filter out envProviderState when it's not requested", () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(
taskData,
newRequestParam({
env: false,
}),
);
it('picks only specific properties for workflow', () => {
const result = builder.buildFromTaskData(taskData);
const result = dataRequestResponseBuilder.build();
expect(result.envProviderState).toStrictEqual({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
});
});
describe('additionalData', () => {
it('picks only specific properties for additional data', () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(taskData, allDataParam);
const result = dataRequestResponseBuilder.build();
expect(result.additionalData).toStrictEqual({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
variables: additionalData.variables,
});
});
});
describe('input data', () => {
const allExceptInputParam = newRequestParam({
input: false,
});
it('drops input data from executeData', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.executeData).toStrictEqual({
node: taskData.executeData!.node,
source: taskData.executeData!.source,
data: {},
});
});
it('drops input data from result', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from result', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from connectionInputData', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.connectionInputData).toStrictEqual([]);
});
});
describe('nodes', () => {
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: ['Code'], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [codeNode.name], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return only DebugHelper's data when only DebugHelper node is requested", () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [debugHelperNode.name], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return DebugHelper's data when only prevNode node is requested", () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [], prevNode: true }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
expect(result.workflow).toStrictEqual({
id: '1',
name: 'Test Workflow',
active: true,
connections: workflow.connectionsBySourceNode,
nodes: [],
pinData: workflow.pinData,
settings: workflow.settings,
staticData: workflow.staticData,
});
});
});

300
packages/cli/src/runners/task-managers/__tests__/data-request-response-stripper.test.ts Normal file
View file

@ -0,0 +1,300 @@
import type { DataRequestResponse, TaskDataRequestParams } from '@n8n/task-runner';
import { mock } from 'jest-mock-extended';
import type { IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import { type INode, type INodeExecutionData } from 'n8n-workflow';
import { DataRequestResponseStripper } from '../data-request-response-stripper';
const triggerNode: INode = mock<INode>({
name: 'Trigger',
});
const debugHelperNode: INode = mock<INode>({
name: 'DebugHelper',
});
const codeNode: INode = mock<INode>({
name: 'Code',
});
const workflow: DataRequestResponse['workflow'] = mock<DataRequestResponse['workflow']>();
const debugHelperNodeOutItems: INodeExecutionData[] = [
{
json: {
uid: 'abb74fd4-bef2-4fae-9d53-ea24e9eb3032',
email: 'Dan.Schmidt31@yahoo.com',
firstname: 'Toni',
lastname: 'Schuster',
password: 'Q!D6C2',
},
pairedItem: {
item: 0,
},
},
];
const codeNodeInputItems: INodeExecutionData[] = debugHelperNodeOutItems;
const envProviderState: DataRequestResponse['envProviderState'] = mock<
DataRequestResponse['envProviderState']
>({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
const additionalData = mock<IWorkflowExecuteAdditionalData>({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
variables: {},
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
/**
* Drawn with https://asciiflow.com/#/
* Task data for an execution of the following WF:
* where ►► denotes the currently being executing node.
* ►►
* ┌───────────┐ ┌─────────────┐ ┌────────┐
* │ Trigger ├──►│ DebugHelper ├───►│ Code │
* └───────────┘ └─────────────┘ └────────┘
*/
const taskData: DataRequestResponse = {
workflow,
inputData: {
main: [codeNodeInputItems],
},
itemIndex: 0,
activeNodeName: codeNode.name,
contextNodeName: codeNode.name,
defaultReturnRunIndex: -1,
mode: 'manual',
envProviderState,
node: codeNode,
runExecutionData: {
startData: {
destinationNode: codeNode.name,
runNodeFilter: [triggerNode.name, debugHelperNode.name, codeNode.name],
},
resultData: {
runData: {
[triggerNode.name]: [
{
hints: [],
startTime: 1730313407328,
executionTime: 1,
source: [],
executionStatus: 'success',
data: {
main: [[]],
},
},
],
[debugHelperNode.name]: [
{
hints: [],
startTime: 1730313407330,
executionTime: 1,
source: [
{
previousNode: triggerNode.name,
},
],
executionStatus: 'success',
data: {
main: [debugHelperNodeOutItems],
},
},
],
},
pinData: {},
},
executionData: {
contextData: {},
nodeExecutionStack: [],
metadata: {},
waitingExecution: {
[codeNode.name]: {
'0': {
main: [codeNodeInputItems],
},
},
},
waitingExecutionSource: {
[codeNode.name]: {
'0': {
main: [
{
previousNode: debugHelperNode.name,
},
],
},
},
},
},
},
runIndex: 0,
selfData: {},
siblingParameters: {},
connectionInputSource: {
main: [
{
previousNode: debugHelperNode.name,
previousNodeOutput: 0,
},
],
},
additionalData,
} as const;
describe('DataRequestResponseStripper', () => {
const allDataParam: TaskDataRequestParams = {
dataOfNodes: 'all',
env: true,
input: true,
prevNode: true,
};
const newRequestParam = (opts: Partial<TaskDataRequestParams>) => ({
...allDataParam,
...opts,
});
describe('all data', () => {
it('should build the runExecutionData as is when everything is requested', () => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(taskData, allDataParam);
const { runExecutionData } = dataRequestResponseBuilder.strip();
expect(runExecutionData).toStrictEqual(taskData.runExecutionData);
});
});
describe('envProviderState', () => {
it("should filter out envProviderState when it's not requested", () => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(
taskData,
newRequestParam({
env: false,
}),
);
const result = dataRequestResponseBuilder.strip();
expect(result.envProviderState).toStrictEqual({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
});
});
describe('input data', () => {
const allExceptInputParam = newRequestParam({
input: false,
});
it('drops input data from result', () => {
const result = new DataRequestResponseStripper(taskData, allExceptInputParam).strip();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from result', () => {
const result = new DataRequestResponseStripper(taskData, allExceptInputParam).strip();
expect(result.inputData).toStrictEqual({});
});
});
describe('nodes', () => {
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: ['Code'], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [codeNode.name], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return only DebugHelper's data when only DebugHelper node is requested", () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [debugHelperNode.name], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return DebugHelper's data when only prevNode node is requested", () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [], prevNode: true }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
});
describe('passthrough properties', () => {
test.each<Array<keyof DataRequestResponse>>([
['workflow'],
['connectionInputSource'],
['node'],
['runIndex'],
['itemIndex'],
['activeNodeName'],
['siblingParameters'],
['mode'],
['defaultReturnRunIndex'],
['selfData'],
['contextNodeName'],
['additionalData'],
])("it doesn't change %s", (propertyName) => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(taskData, allDataParam);
const result = dataRequestResponseBuilder.strip();
expect(result[propertyName]).toBe(taskData[propertyName]);
});
});
});

190
packages/cli/src/runners/task-managers/data-request-response-builder.ts
View file

@ -1,63 +1,30 @@
import type {
DataRequestResponse,
BrokerMessage,
PartialAdditionalData,
TaskData,
} from '@n8n/task-runner';
import type {
EnvProviderState,
IExecuteData,
INodeExecutionData,
IPinData,
IRunData,
IRunExecutionData,
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
Workflow,
WorkflowParameters,
} from 'n8n-workflow';
import type { DataRequestResponse, PartialAdditionalData, TaskData } from '@n8n/task-runner';
import type { IWorkflowExecuteAdditionalData, Workflow, WorkflowParameters } from 'n8n-workflow';
/**
* Builds the response to a data request coming from a Task Runner. Tries to minimize
* the amount of data that is sent to the runner by only providing what is requested.
* Transforms TaskData to DataRequestResponse. The main purpose of the
* transformation is to make sure there is no duplication in the data
* (e.g. connectionInputData and executeData.data can be derived from
* inputData).
*/
export class DataRequestResponseBuilder {
private requestedNodeNames = new Set<string>();
constructor(
private readonly taskData: TaskData,
private readonly requestParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'],
) {
this.requestedNodeNames = new Set(requestParams.dataOfNodes);
if (this.requestParams.prevNode && this.requestParams.dataOfNodes !== 'all') {
this.requestedNodeNames.add(this.determinePrevNodeName());
}
}
/**
* Builds a response to the data request
*/
build(): DataRequestResponse {
const { taskData: td } = this;
buildFromTaskData(taskData: TaskData): DataRequestResponse {
return {
workflow: this.buildWorkflow(td.workflow),
connectionInputData: this.buildConnectionInputData(td.connectionInputData),
inputData: this.buildInputData(td.inputData),
itemIndex: td.itemIndex,
activeNodeName: td.activeNodeName,
contextNodeName: td.contextNodeName,
defaultReturnRunIndex: td.defaultReturnRunIndex,
mode: td.mode,
envProviderState: this.buildEnvProviderState(td.envProviderState),
node: td.node, // The current node being executed
runExecutionData: this.buildRunExecutionData(td.runExecutionData),
runIndex: td.runIndex,
selfData: td.selfData,
siblingParameters: td.siblingParameters,
executeData: this.buildExecuteData(td.executeData),
additionalData: this.buildAdditionalData(td.additionalData),
workflow: this.buildWorkflow(taskData.workflow),
inputData: taskData.inputData,
connectionInputSource: taskData.executeData?.source ?? null,
itemIndex: taskData.itemIndex,
activeNodeName: taskData.activeNodeName,
contextNodeName: taskData.contextNodeName,
defaultReturnRunIndex: taskData.defaultReturnRunIndex,
mode: taskData.mode,
envProviderState: taskData.envProviderState,
node: taskData.node,
runExecutionData: taskData.runExecutionData,
runIndex: taskData.runIndex,
selfData: taskData.selfData,
siblingParameters: taskData.siblingParameters,
additionalData: this.buildAdditionalData(taskData.additionalData),
};
}
@ -80,86 +47,6 @@ export class DataRequestResponseBuilder {
};
}
private buildExecuteData(executeData: IExecuteData | undefined): IExecuteData | undefined {
if (executeData === undefined) {
return undefined;
}
return {
node: executeData.node, // The current node being executed
data: this.requestParams.input ? executeData.data : {},
source: executeData.source,
};
}
private buildRunExecutionData(runExecutionData: IRunExecutionData): IRunExecutionData {
if (this.requestParams.dataOfNodes === 'all') {
return runExecutionData;
}
return {
startData: runExecutionData.startData,
resultData: {
error: runExecutionData.resultData.error,
lastNodeExecuted: runExecutionData.resultData.lastNodeExecuted,
metadata: runExecutionData.resultData.metadata,
runData: this.buildRunData(runExecutionData.resultData.runData),
pinData: this.buildPinData(runExecutionData.resultData.pinData),
},
executionData: runExecutionData.executionData
? {
// TODO: Figure out what these two are and can they be filtered
contextData: runExecutionData.executionData?.contextData,
nodeExecutionStack: runExecutionData.executionData.nodeExecutionStack,
metadata: runExecutionData.executionData.metadata,
waitingExecution: runExecutionData.executionData.waitingExecution,
waitingExecutionSource: runExecutionData.executionData.waitingExecutionSource,
}
: undefined,
};
}
private buildRunData(runData: IRunData): IRunData {
return this.filterObjectByNodeNames(runData);
}
private buildPinData(pinData: IPinData | undefined): IPinData | undefined {
return pinData ? this.filterObjectByNodeNames(pinData) : undefined;
}
private buildEnvProviderState(envProviderState: EnvProviderState): EnvProviderState {
if (this.requestParams.env) {
// In case `isEnvAccessBlocked` = true, the provider state has already sanitized
// the environment variables and we can return it as is.
return envProviderState;
}
return {
env: {},
isEnvAccessBlocked: envProviderState.isEnvAccessBlocked,
isProcessAvailable: envProviderState.isProcessAvailable,
};
}
private buildInputData(inputData: ITaskDataConnections): ITaskDataConnections {
if (this.requestParams.input) {
return inputData;
}
return {};
}
private buildConnectionInputData(
connectionInputData: INodeExecutionData[],
): INodeExecutionData[] {
if (this.requestParams.input) {
return connectionInputData;
}
return [];
}
private buildWorkflow(workflow: Workflow): Omit<WorkflowParameters, 'nodeTypes'> {
return {
id: workflow.id,
@ -172,37 +59,4 @@ export class DataRequestResponseBuilder {
staticData: workflow.staticData,
};
}
/**
* Assuming the given `obj` is an object where the keys are node names,
* filters the object to only include the node names that are requested.
*/
private filterObjectByNodeNames<T extends Record<string, unknown>>(obj: T): T {
if (this.requestParams.dataOfNodes === 'all') {
return obj;
}
const filteredObj: T = {} as T;
for (const nodeName in obj) {
if (!Object.prototype.hasOwnProperty.call(obj, nodeName)) {
continue;
}
if (this.requestedNodeNames.has(nodeName)) {
filteredObj[nodeName] = obj[nodeName];
}
}
return filteredObj;
}
private determinePrevNodeName(): string {
const sourceData = this.taskData.executeData?.source?.main?.[0];
if (!sourceData) {
return '';
}
return sourceData.previousNode;
}
}

131
packages/cli/src/runners/task-managers/data-request-response-stripper.ts Normal file
View file

@ -0,0 +1,131 @@
import type { DataRequestResponse, BrokerMessage } from '@n8n/task-runner';
import type {
EnvProviderState,
IPinData,
IRunData,
IRunExecutionData,
ITaskDataConnections,
} from 'n8n-workflow';
/**
* Strips data from data request response based on the specified parameters
*/
export class DataRequestResponseStripper {
private requestedNodeNames = new Set<string>();
constructor(
private readonly dataResponse: DataRequestResponse,
private readonly stripParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'],
) {
this.requestedNodeNames = new Set(stripParams.dataOfNodes);
if (this.stripParams.prevNode && this.stripParams.dataOfNodes !== 'all') {
this.requestedNodeNames.add(this.determinePrevNodeName());
}
}
/**
* Builds a response to the data request
*/
strip(): DataRequestResponse {
const { dataResponse: dr } = this;
return {
...dr,
inputData: this.stripInputData(dr.inputData),
envProviderState: this.stripEnvProviderState(dr.envProviderState),
runExecutionData: this.stripRunExecutionData(dr.runExecutionData),
};
}
private stripRunExecutionData(runExecutionData: IRunExecutionData): IRunExecutionData {
if (this.stripParams.dataOfNodes === 'all') {
return runExecutionData;
}
return {
startData: runExecutionData.startData,
resultData: {
error: runExecutionData.resultData.error,
lastNodeExecuted: runExecutionData.resultData.lastNodeExecuted,
metadata: runExecutionData.resultData.metadata,
runData: this.stripRunData(runExecutionData.resultData.runData),
pinData: this.stripPinData(runExecutionData.resultData.pinData),
},
executionData: runExecutionData.executionData
? {
// TODO: Figure out what these two are and can they be stripped
contextData: runExecutionData.executionData?.contextData,
nodeExecutionStack: runExecutionData.executionData.nodeExecutionStack,
metadata: runExecutionData.executionData.metadata,
waitingExecution: runExecutionData.executionData.waitingExecution,
waitingExecutionSource: runExecutionData.executionData.waitingExecutionSource,
}
: undefined,
};
}
private stripRunData(runData: IRunData): IRunData {
return this.filterObjectByNodeNames(runData);
}
private stripPinData(pinData: IPinData | undefined): IPinData | undefined {
return pinData ? this.filterObjectByNodeNames(pinData) : undefined;
}
private stripEnvProviderState(envProviderState: EnvProviderState): EnvProviderState {
if (this.stripParams.env) {
// In case `isEnvAccessBlocked` = true, the provider state has already sanitized
// the environment variables and we can return it as is.
return envProviderState;
}
return {
env: {},
isEnvAccessBlocked: envProviderState.isEnvAccessBlocked,
isProcessAvailable: envProviderState.isProcessAvailable,
};
}
private stripInputData(inputData: ITaskDataConnections): ITaskDataConnections {
if (this.stripParams.input) {
return inputData;
}
return {};
}
/**
* Assuming the given `obj` is an object where the keys are node names,
* filters the object to only include the node names that are requested.
*/
private filterObjectByNodeNames<T extends Record<string, unknown>>(obj: T): T {
if (this.stripParams.dataOfNodes === 'all') {
return obj;
}
const filteredObj: T = {} as T;
for (const nodeName in obj) {
if (!Object.prototype.hasOwnProperty.call(obj, nodeName)) {
continue;
}
if (this.requestedNodeNames.has(nodeName)) {
filteredObj[nodeName] = obj[nodeName];
}
}
return filteredObj;
}
private determinePrevNodeName(): string {
const sourceData = this.dataResponse.connectionInputSource?.main?.[0];
if (!sourceData) {
return '';
}
return sourceData.previousNode;
}
}

33
packages/cli/src/runners/task-managers/task-manager.ts
View file

@ -1,5 +1,6 @@
import { TaskRunnersConfig } from '@n8n/config';
import type { TaskResultData, RequesterMessage, BrokerMessage, TaskData } from '@n8n/task-runner';
import { RPC_ALLOW_LIST } from '@n8n/task-runner';
import { DataRequestResponseReconstruct, RPC_ALLOW_LIST } from '@n8n/task-runner';
import type {
EnvProviderState,
IExecuteFunctions,
@ -17,11 +18,13 @@ import type {
} from 'n8n-workflow';
import { createResultOk, createResultError } from 'n8n-workflow';
import { nanoid } from 'nanoid';
import { Service } from 'typedi';
import * as a from 'node:assert/strict';
import Container, { Service } from 'typedi';
import { NodeTypes } from '@/node-types';
import { DataRequestResponseBuilder } from './data-request-response-builder';
import { DataRequestResponseStripper } from './data-request-response-stripper';
export type RequestAccept = (jobId: string) => void;
export type RequestReject = (reason: string) => void;
@ -56,6 +59,10 @@ export abstract class TaskManager {
tasks: Map<string, Task> = new Map();
private readonly runnerConfig = Container.get(TaskRunnersConfig);
private readonly dataResponseBuilder = new DataRequestResponseBuilder();
constructor(private readonly nodeTypes: NodeTypes) {}
async startTask<TData, TError>(
@ -237,14 +244,30 @@ export abstract class TaskManager {
return;
}
const dataRequestResponseBuilder = new DataRequestResponseBuilder(job.data, requestParams);
const requestedData = dataRequestResponseBuilder.build();
const dataRequestResponse = this.dataResponseBuilder.buildFromTaskData(job.data);
if (this.runnerConfig.assertDeduplicationOutput) {
const reconstruct = new DataRequestResponseReconstruct();
a.deepStrictEqual(
reconstruct.reconstructConnectionInputData(dataRequestResponse.inputData),
job.data.connectionInputData,
);
a.deepStrictEqual(
reconstruct.reconstructExecuteData(dataRequestResponse),
job.data.executeData,
);
}
const strippedData = new DataRequestResponseStripper(
dataRequestResponse,
requestParams,
).strip();
this.sendMessage({
type: 'requester:taskdataresponse',
taskId,
requestId,
data: requestedData,
data: strippedData,
});
}

2
packages/core/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-core",
"version": "1.66.0",
"version": "1.67.0",
"description": "Core functionality of n8n",
"main": "dist/index",
"types": "dist/index.d.ts",

2
packages/design-system/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-design-system",
"version": "1.56.0",
"version": "1.57.0",
"main": "src/main.ts",
"import": "src/main.ts",
"scripts": {

2
packages/editor-ui/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-editor-ui",
"version": "1.66.0",
"version": "1.67.0",
"description": "Workflow Editor UI for n8n",
"main": "index.js",
"scripts": {

154
packages/editor-ui/src/components/Error/NodeErrorView.test.ts
View file

@ -1,38 +1,61 @@
import { createComponentRenderer } from '@/__tests__/render';
import { SETTINGS_STORE_DEFAULT_STATE } from '@/__tests__/utils';
import NodeErrorView from '@/components/Error/NodeErrorView.vue';
import { STORES } from '@/constants';
import { createTestingPinia } from '@pinia/testing';
import { type INode } from 'n8n-workflow';
import type { NodeError } from 'n8n-workflow';
import { useAssistantStore } from '@/stores/assistant.store';
import { useNodeTypesStore } from '@/stores/nodeTypes.store';
import { mockedStore } from '@/__tests__/utils';
import userEvent from '@testing-library/user-event';
import { useNDVStore } from '@/stores/ndv.store';
const DEFAULT_SETUP = {
pinia: createTestingPinia({
initialState: {
[STORES.SETTINGS]: SETTINGS_STORE_DEFAULT_STATE,
},
}),
};
const renderComponent = createComponentRenderer(NodeErrorView);
const renderComponent = createComponentRenderer(NodeErrorView, DEFAULT_SETUP);
let mockAiAssistantStore: ReturnType<typeof mockedStore<typeof useAssistantStore>>;
let mockNodeTypeStore: ReturnType<typeof mockedStore<typeof useNodeTypesStore>>;
let mockNdvStore: ReturnType<typeof mockedStore<typeof useNDVStore>>;
describe('NodeErrorView.vue', () => {
let mockNode: INode;
afterEach(() => {
mockNode = {
parameters: {
mode: 'runOnceForAllItems',
language: 'javaScript',
jsCode: 'cons error = 9;',
notice: '',
let error: NodeError;
beforeEach(() => {
createTestingPinia();
mockAiAssistantStore = mockedStore(useAssistantStore);
mockNodeTypeStore = mockedStore(useNodeTypesStore);
mockNdvStore = mockedStore(useNDVStore);
//@ts-expect-error
error = {
name: 'NodeOperationError',
message: 'Test error message',
description: 'Test error description',
context: {
descriptionKey: 'noInputConnection',
nodeCause: 'Test node cause',
runIndex: '1',
itemIndex: '2',
parameter: 'testParameter',
data: { key: 'value' },
causeDetailed: 'Detailed cause',
},
id: 'd1ce5dc9-f9ae-4ac6-84e5-0696ba175dd9',
name: 'Code',
type: 'n8n-nodes-base.code',
typeVersion: 2,
position: [940, 240],
node: {
parameters: {
mode: 'runOnceForAllItems',
language: 'javaScript',
jsCode: 'cons error = 9;',
notice: '',
},
id: 'd1ce5dc9-f9ae-4ac6-84e5-0696ba175dd9',
name: 'ErrorCode',
type: 'n8n-nodes-base.code',
typeVersion: 2,
position: [940, 240],
},
stack: 'Test stack trace',
};
});
afterEach(() => {
vi.clearAllMocks();
});
@ -40,7 +63,7 @@ describe('NodeErrorView.vue', () => {
const { getByTestId } = renderComponent({
props: {
error: {
node: mockNode,
node: error.node,
messages: ['Unexpected identifier [line 1]'],
},
},
@ -55,7 +78,7 @@ describe('NodeErrorView.vue', () => {
const { getByTestId } = renderComponent({
props: {
error: {
node: mockNode,
node: error.node,
message: 'Unexpected identifier [line 1]',
},
},
@ -67,24 +90,20 @@ describe('NodeErrorView.vue', () => {
});
it('should not render AI assistant button when error happens in deprecated function node', async () => {
const aiAssistantStore = useAssistantStore(DEFAULT_SETUP.pinia);
const nodeTypeStore = useNodeTypesStore(DEFAULT_SETUP.pinia);
//@ts-expect-error
nodeTypeStore.getNodeType = vi.fn(() => ({
mockNodeTypeStore.getNodeType = vi.fn(() => ({
type: 'n8n-nodes-base.function',
typeVersion: 1,
hidden: true,
}));
//@ts-expect-error
aiAssistantStore.canShowAssistantButtonsOnCanvas = true;
mockAiAssistantStore.canShowAssistantButtonsOnCanvas = true;
const { queryByTestId } = renderComponent({
props: {
error: {
node: {
...mockNode,
...error.node,
type: 'n8n-nodes-base.function',
typeVersion: 1,
},
@ -96,4 +115,73 @@ describe('NodeErrorView.vue', () => {
expect(aiAssistantButton).toBeNull();
});
it('renders error message', () => {
const { getByTestId } = renderComponent({
props: { error },
});
expect(getByTestId('node-error-message').textContent).toContain('Test error message');
});
it('renders error description', () => {
const { getByTestId } = renderComponent({
props: { error },
});
expect(getByTestId('node-error-description').innerHTML).toContain(
'This node has no input data. Please make sure this node is connected to another node.',
);
});
it('renders stack trace', () => {
const { getByText } = renderComponent({
props: { error },
});
expect(getByText('Test stack trace')).toBeTruthy();
});
it('renders open node button when the error is in sub node', () => {
const { getByTestId, queryByTestId } = renderComponent({
props: {
error: {
...error,
name: 'NodeOperationError',
functionality: 'configuration-node',
},
},
});
expect(getByTestId('node-error-view-open-node-button')).toHaveTextContent('Open errored node');
expect(queryByTestId('ask-assistant-button')).not.toBeInTheDocument();
});
it('does not renders open node button when the error is in sub node', () => {
mockAiAssistantStore.canShowAssistantButtonsOnCanvas = true;
const { getByTestId, queryByTestId } = renderComponent({
props: {
error,
},
});
expect(queryByTestId('node-error-view-open-node-button')).not.toBeInTheDocument();
expect(getByTestId('ask-assistant-button')).toBeInTheDocument();
});
it('open error node details when open error node is clicked', async () => {
const { getByTestId, emitted } = renderComponent({
props: {
error: {
...error,
name: 'NodeOperationError',
functionality: 'configuration-node',
},
},
});
await userEvent.click(getByTestId('node-error-view-open-node-button'));
expect(emitted().click).toHaveLength(1);
expect(mockNdvStore.activeNodeName).toBe(error.node.name);
});
});

55
packages/editor-ui/src/components/Error/NodeErrorView.vue
View file

@ -117,7 +117,7 @@ const prepareRawMessages = computed(() => {
});
const isAskAssistantAvailable = computed(() => {
if (!node.value) {
if (!node.value || isSubNodeError.value) {
return false;
}
const isCustomNode = node.value.type === undefined || isCommunityPackageName(node.value.type);
@ -132,6 +132,13 @@ const assistantAlreadyAsked = computed(() => {
});
});
const isSubNodeError = computed(() => {
return (
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node'
);
});
function nodeVersionTag(nodeType: NodeError['node']): string {
if (!nodeType || ('hidden' in nodeType && nodeType.hidden)) {
return i18n.baseText('nodeSettings.deprecated');
@ -153,19 +160,6 @@ function prepareDescription(description: string): string {
}
function getErrorDescription(): string {
const isSubNodeError =
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node';
if (isSubNodeError) {
return prepareDescription(
props.error.description +
i18n.baseText('pushConnection.executionError.openNode', {
interpolate: { node: props.error.node.name },
}),
);
}
if (props.error.context?.descriptionKey) {
const interpolate = {
nodeCause: props.error.context.nodeCause as string,
@ -205,13 +199,10 @@ function addItemIndexSuffix(message: string): string {
function getErrorMessage(): string {
let message = '';
const isSubNodeError =
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node';
const isNonEmptyString = (value?: unknown): value is string =>
!!value && typeof value === 'string';
if (isSubNodeError) {
if (isSubNodeError.value) {
message = i18n.baseText('nodeErrorView.errorSubNode', {
interpolate: { node: props.error.node.name },
});
@ -390,6 +381,10 @@ function nodeIsHidden() {
return nodeType?.hidden ?? false;
}
const onOpenErrorNodeDetailClick = () => {
ndvStore.activeNodeName = props.error.node.name;
};
async function onAskAssistantClick() {
const { message, lineNumber, description } = props.error;
const sessionInProgress = !assistantStore.isSessionEnded;
@ -428,14 +423,25 @@ async function onAskAssistantClick() {
</div>
</div>
<div
v-if="error.description || error.context?.descriptionKey"
v-if="(error.description || error.context?.descriptionKey) && !isSubNodeError"
data-test-id="node-error-description"
class="node-error-view__header-description"
v-n8n-html="getErrorDescription()"
></div>
<div v-if="isSubNodeError">
<n8n-button
icon="arrow-right"
type="secondary"
:label="i18n.baseText('pushConnection.executionError.openNode')"
class="node-error-view__button"
data-test-id="node-error-view-open-node-button"
@click="onOpenErrorNodeDetailClick"
/>
</div>
<div
v-if="isAskAssistantAvailable"
class="node-error-view__assistant-button"
class="node-error-view__button"
data-test-id="node-error-view-ask-assistant-button"
>
<InlineAskAssistantButton :asked="assistantAlreadyAsked" @click="onAskAssistantClick" />
@ -696,9 +702,14 @@ async function onAskAssistantClick() {
}
}
&__assistant-button {
&__button {
margin-left: var(--spacing-s);
margin-bottom: var(--spacing-xs);
flex-direction: row-reverse;
span {
margin-right: var(--spacing-5xs);
margin-left: var(--spacing-5xs);
}
}
&__debugging {
@ -831,7 +842,7 @@ async function onAskAssistantClick() {
}
}
.node-error-view__assistant-button {
.node-error-view__button {
margin-top: var(--spacing-xs);
}
</style>

6
packages/editor-ui/src/components/RunData.vue
View file

@ -289,7 +289,7 @@ export default defineComponent({
return false;
}
const canPinNode = usePinnedData(this.node).canPinNode(false);
const canPinNode = usePinnedData(this.node).canPinNode(false, this.currentOutputIndex);
return (
canPinNode &&
@ -1214,9 +1214,7 @@ export default defineComponent({
<template>
<div :class="['run-data', $style.container]" @mouseover="activatePane">
<n8n-callout
v-if="
canPinData && pinnedData.hasData.value && !editMode.enabled && !isProductionExecutionPreview
"
v-if="pinnedData.hasData.value && !editMode.enabled && !isProductionExecutionPreview"
theme="secondary"
icon="thumbtack"
:class="$style.pinnedDataCallout"

657
packages/editor-ui/src/components/SettingsLogStreaming/EventDestinationSettingsModal.ee.vue
View file

@ -1,17 +1,15 @@
<script lang="ts">
<script setup lang="ts">
import { computed, onMounted, ref } from 'vue';
import { get, set, unset } from 'lodash-es';
import { mapStores } from 'pinia';
import { useLogStreamingStore } from '@/stores/logStreaming.store';
import { useNDVStore } from '@/stores/ndv.store';
import { useWorkflowsStore } from '@/stores/workflows.store';
import ParameterInputList from '@/components/ParameterInputList.vue';
import type { IMenuItem, INodeUi, IUpdateInformation, ModalKey } from '@/Interface';
import type {
IDataObject,
NodeParameterValue,
MessageEventBusDestinationOptions,
INodeParameters,
NodeParameterValueType,
MessageEventBusDestinationSentryOptions,
MessageEventBusDestinationSyslogOptions,
MessageEventBusDestinationWebhookOptions,
} from 'n8n-workflow';
import {
deepCopy,
@ -22,338 +20,337 @@ import {
defaultMessageEventBusDestinationSyslogOptions,
defaultMessageEventBusDestinationSentryOptions,
} from 'n8n-workflow';
import type { PropType } from 'vue';
import { defineComponent } from 'vue';
import type { EventBus } from 'n8n-design-system';
import { createEventBus } from 'n8n-design-system/utils';
import { useLogStreamingStore } from '@/stores/logStreaming.store';
import { useNDVStore } from '@/stores/ndv.store';
import { useWorkflowsStore } from '@/stores/workflows.store';
import ParameterInputList from '@/components/ParameterInputList.vue';
import type { IMenuItem, IUpdateInformation, ModalKey } from '@/Interface';
import { LOG_STREAM_MODAL_KEY, MODAL_CONFIRM } from '@/constants';
import Modal from '@/components/Modal.vue';
import { useI18n } from '@/composables/useI18n';
import { useMessage } from '@/composables/useMessage';
import { useUIStore } from '@/stores/ui.store';
import { hasPermission } from '@/utils/rbac/permissions';
import { destinationToFakeINodeUi } from '@/components/SettingsLogStreaming/Helpers.ee';
import type { BaseTextKey } from '@/plugins/i18n';
import InlineNameEdit from '@/components/InlineNameEdit.vue';
import SaveButton from '@/components/SaveButton.vue';
import EventSelection from '@/components/SettingsLogStreaming/EventSelection.ee.vue';
import { useTelemetry } from '@/composables/useTelemetry';
import { useRootStore } from '@/stores/root.store';
import {
webhookModalDescription,
sentryModalDescription,
syslogModalDescription,
} from './descriptions.ee';
import type { BaseTextKey } from '@/plugins/i18n';
import InlineNameEdit from '@/components/InlineNameEdit.vue';
import SaveButton from '@/components/SaveButton.vue';
import EventSelection from '@/components/SettingsLogStreaming/EventSelection.ee.vue';
import type { EventBus } from 'n8n-design-system';
import { createEventBus } from 'n8n-design-system/utils';
import { useTelemetry } from '@/composables/useTelemetry';
import { useRootStore } from '@/stores/root.store';
export default defineComponent({
name: 'EventDestinationSettingsModal',
components: {
Modal,
ParameterInputList,
InlineNameEdit,
SaveButton,
EventSelection,
defineOptions({ name: 'EventDestinationSettingsModal' });
const props = withDefaults(
defineProps<{
modalName: ModalKey;
destination?: MessageEventBusDestinationOptions;
isNew?: boolean;
eventBus?: EventBus;
}>(),
{
destination: () => deepCopy(defaultMessageEventBusDestinationOptions),
isNew: false,
},
props: {
modalName: {
type: String as PropType<ModalKey>,
required: true,
},
destination: {
type: Object,
default: () => deepCopy(defaultMessageEventBusDestinationOptions),
},
isNew: Boolean,
eventBus: {
type: Object as PropType<EventBus>,
},
},
setup() {
return {
...useMessage(),
};
},
data() {
return {
unchanged: !this.isNew,
activeTab: 'settings',
hasOnceBeenSaved: !this.isNew,
isSaving: false,
isDeleting: false,
loading: false,
showRemoveConfirm: false,
typeSelectValue: '',
typeSelectPlaceholder: 'Destination Type',
nodeParameters: deepCopy(defaultMessageEventBusDestinationOptions) as INodeParameters,
webhookDescription: webhookModalDescription,
sentryDescription: sentryModalDescription,
syslogDescription: syslogModalDescription,
modalBus: createEventBus(),
headerLabel: this.destination.label,
testMessageSent: false,
testMessageResult: false,
LOG_STREAM_MODAL_KEY,
};
},
computed: {
...mapStores(useUIStore, useLogStreamingStore, useNDVStore, useWorkflowsStore),
typeSelectOptions(): Array<{ value: string; label: BaseTextKey }> {
const options: Array<{ value: string; label: BaseTextKey }> = [];
for (const t of messageEventBusDestinationTypeNames) {
if (t === MessageEventBusDestinationTypeNames.abstract) {
continue;
}
options.push({
value: t,
label: `settings.log-streaming.${t}` as BaseTextKey,
});
}
return options;
},
isTypeAbstract(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.abstract;
},
isTypeWebhook(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.webhook;
},
isTypeSyslog(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.syslog;
},
isTypeSentry(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.sentry;
},
node(): INodeUi {
return destinationToFakeINodeUi(this.nodeParameters);
},
typeLabelName(): BaseTextKey {
return `settings.log-streaming.${this.nodeParameters.__type}` as BaseTextKey;
},
sidebarItems(): IMenuItem[] {
const items: IMenuItem[] = [
{
id: 'settings',
label: this.$locale.baseText('settings.log-streaming.tab.settings'),
position: 'top',
},
];
if (!this.isTypeAbstract) {
items.push({
id: 'events',
label: this.$locale.baseText('settings.log-streaming.tab.events'),
position: 'top',
});
}
return items;
},
canManageLogStreaming(): boolean {
return hasPermission(['rbac'], { rbac: { scope: 'logStreaming:manage' } });
},
},
mounted() {
this.setupNode(
Object.assign(deepCopy(defaultMessageEventBusDestinationOptions), this.destination),
);
this.workflowsStore.$onAction(
({
name, // name of the action
args, // array of parameters passed to the action
}) => {
if (name === 'updateNodeProperties') {
for (const arg of args) {
if (arg.name === this.destination.id) {
if ('credentials' in arg.properties) {
this.unchanged = false;
this.nodeParameters.credentials = arg.properties
.credentials as NodeParameterValueType;
}
}
}
}
},
);
},
methods: {
onInput() {
this.unchanged = false;
this.testMessageSent = false;
},
onTabSelect(tab: string) {
this.activeTab = tab;
},
onLabelChange(value: string) {
this.onInput();
this.headerLabel = value;
this.nodeParameters.label = value;
},
setupNode(options: MessageEventBusDestinationOptions) {
this.workflowsStore.removeNode(this.node);
this.ndvStore.activeNodeName = options.id ?? 'thisshouldnothappen';
this.workflowsStore.addNode(destinationToFakeINodeUi(options));
this.nodeParameters = options as INodeParameters;
this.logStreamingStore.items[this.destination.id].destination = options;
},
onTypeSelectInput(destinationType: MessageEventBusDestinationTypeNames) {
this.typeSelectValue = destinationType;
},
async onContinueAddClicked() {
let newDestination;
switch (this.typeSelectValue) {
case MessageEventBusDestinationTypeNames.syslog:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSyslogOptions), {
id: this.destination.id,
});
break;
case MessageEventBusDestinationTypeNames.sentry:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSentryOptions), {
id: this.destination.id,
});
break;
case MessageEventBusDestinationTypeNames.webhook:
newDestination = Object.assign(
deepCopy(defaultMessageEventBusDestinationWebhookOptions),
{ id: this.destination.id },
);
break;
}
);
const { modalName, destination, isNew, eventBus } = props;
if (newDestination) {
this.headerLabel = newDestination?.label ?? this.headerLabel;
this.setupNode(newDestination);
}
},
valueChanged(parameterData: IUpdateInformation) {
this.unchanged = false;
this.testMessageSent = false;
const newValue: NodeParameterValue = parameterData.value as string | number;
const parameterPath = parameterData.name?.startsWith('parameters.')
? parameterData.name.split('.').slice(1).join('.')
: parameterData.name || '';
const i18n = useI18n();
const { confirm } = useMessage();
const telemetry = useTelemetry();
const logStreamingStore = useLogStreamingStore();
const ndvStore = useNDVStore();
const workflowsStore = useWorkflowsStore();
const uiStore = useUIStore();
const nodeParameters = deepCopy(this.nodeParameters);
const unchanged = ref(!isNew);
const activeTab = ref('settings');
const hasOnceBeenSaved = ref(!isNew);
const isSaving = ref(false);
const isDeleting = ref(false);
const loading = ref(false);
const typeSelectValue = ref('');
const typeSelectPlaceholder = ref('Destination Type');
const nodeParameters = ref(deepCopy(defaultMessageEventBusDestinationOptions) as INodeParameters);
const webhookDescription = ref(webhookModalDescription);
const sentryDescription = ref(sentryModalDescription);
const syslogDescription = ref(syslogModalDescription);
const modalBus = ref(createEventBus());
const headerLabel = ref(destination.label!);
const testMessageSent = ref(false);
const testMessageResult = ref(false);
// Check if the path is supposed to change an array and if so get
// the needed data like path and index
const parameterPathArray = parameterPath.match(/(.*)\[(\d+)\]$/);
// Apply the new value
if (parameterData.value === undefined && parameterPathArray !== null) {
// Delete array item
const path = parameterPathArray[1] as keyof MessageEventBusDestinationOptions;
const index = parameterPathArray[2];
const data = get(nodeParameters, path);
if (Array.isArray(data)) {
data.splice(parseInt(index, 10), 1);
nodeParameters[path] = data as never;
}
} else {
if (newValue === undefined) {
unset(nodeParameters, parameterPath);
} else {
set(nodeParameters, parameterPath, newValue);
}
}
this.nodeParameters = deepCopy(nodeParameters);
this.workflowsStore.updateNodeProperties({
name: this.node.name,
properties: { parameters: this.nodeParameters as unknown as IDataObject, position: [0, 0] },
});
if (this.hasOnceBeenSaved) {
this.logStreamingStore.updateDestination(this.nodeParameters);
}
},
async sendTestEvent() {
this.testMessageResult = await this.logStreamingStore.sendTestMessage(this.nodeParameters);
this.testMessageSent = true;
},
async removeThis() {
const deleteConfirmed = await this.confirm(
this.$locale.baseText('settings.log-streaming.destinationDelete.message', {
interpolate: { destinationName: this.destination.label },
}),
this.$locale.baseText('settings.log-streaming.destinationDelete.headline'),
{
type: 'warning',
confirmButtonText: this.$locale.baseText(
'settings.log-streaming.destinationDelete.confirmButtonText',
),
cancelButtonText: this.$locale.baseText(
'settings.log-streaming.destinationDelete.cancelButtonText',
),
},
);
if (deleteConfirmed !== MODAL_CONFIRM) {
return;
} else {
this.callEventBus('remove', this.destination.id);
this.uiStore.closeModal(LOG_STREAM_MODAL_KEY);
this.uiStore.stateIsDirty = false;
}
},
onModalClose() {
if (!this.hasOnceBeenSaved) {
this.workflowsStore.removeNode(this.node);
if (this.nodeParameters.id && typeof this.nodeParameters.id !== 'object') {
this.logStreamingStore.removeDestination(this.nodeParameters.id.toString());
}
}
this.ndvStore.activeNodeName = null;
this.callEventBus('closing', this.destination.id);
this.uiStore.stateIsDirty = false;
},
async saveDestination() {
if (this.unchanged || !this.destination.id) {
return;
}
const saveResult = await this.logStreamingStore.saveDestination(this.nodeParameters);
if (saveResult) {
this.hasOnceBeenSaved = true;
this.testMessageSent = false;
this.unchanged = true;
this.callEventBus('destinationWasSaved', this.destination.id);
this.uiStore.stateIsDirty = false;
const destinationType = (
this.nodeParameters.__type && typeof this.nodeParameters.__type !== 'object'
? `${this.nodeParameters.__type}`
: 'unknown'
)
.replace('$$MessageEventBusDestination', '')
.toLowerCase();
const isComplete = () => {
if (this.isTypeWebhook) {
return this.destination.host !== '';
} else if (this.isTypeSentry) {
return this.destination.dsn !== '';
} else if (this.isTypeSyslog) {
return (
this.destination.host !== '' &&
this.destination.port !== undefined &&
this.destination.protocol !== '' &&
this.destination.facility !== undefined &&
this.destination.app_name !== ''
);
}
return false;
};
useTelemetry().track('User updated log streaming destination', {
instance_id: useRootStore().instanceId,
destination_type: destinationType,
is_complete: isComplete(),
is_active: this.destination.enabled,
});
}
},
callEventBus(event: string, data: unknown) {
if (this.eventBus) {
this.eventBus.emit(event, data);
}
},
},
const typeSelectOptions = computed(() => {
const options: Array<{ value: string; label: BaseTextKey }> = [];
for (const t of messageEventBusDestinationTypeNames) {
if (t === MessageEventBusDestinationTypeNames.abstract) {
continue;
}
options.push({
value: t,
label: `settings.log-streaming.${t}` as BaseTextKey,
});
}
return options;
});
const isTypeAbstract = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.abstract,
);
const isTypeWebhook = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.webhook,
);
const isTypeSyslog = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.syslog,
);
const isTypeSentry = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.sentry,
);
const node = computed(() => destinationToFakeINodeUi(nodeParameters.value));
const typeLabelName = computed(
() => `settings.log-streaming.${nodeParameters.value.__type}` as BaseTextKey,
);
const sidebarItems = computed(() => {
const items: IMenuItem[] = [
{
id: 'settings',
label: i18n.baseText('settings.log-streaming.tab.settings'),
position: 'top',
},
];
if (!isTypeAbstract.value) {
items.push({
id: 'events',
label: i18n.baseText('settings.log-streaming.tab.events'),
position: 'top',
});
}
return items;
});
const canManageLogStreaming = computed(() =>
hasPermission(['rbac'], { rbac: { scope: 'logStreaming:manage' } }),
);
onMounted(() => {
setupNode(Object.assign(deepCopy(defaultMessageEventBusDestinationOptions), destination));
workflowsStore.$onAction(({ name, args }) => {
if (name === 'updateNodeProperties') {
for (const arg of args) {
if (arg.name === destination.id) {
if ('credentials' in arg.properties) {
unchanged.value = false;
nodeParameters.value.credentials = arg.properties.credentials as NodeParameterValueType;
}
}
}
}
});
});
function onInput() {
unchanged.value = false;
testMessageSent.value = false;
}
function onTabSelect(tab: string) {
activeTab.value = tab;
}
function onLabelChange(value: string) {
onInput();
headerLabel.value = value;
nodeParameters.value.label = value;
}
function setupNode(options: MessageEventBusDestinationOptions) {
workflowsStore.removeNode(node.value);
ndvStore.activeNodeName = options.id ?? 'thisshouldnothappen';
workflowsStore.addNode(destinationToFakeINodeUi(options));
nodeParameters.value = options as INodeParameters;
logStreamingStore.items[destination.id!].destination = options;
}
function onTypeSelectInput(destinationType: MessageEventBusDestinationTypeNames) {
typeSelectValue.value = destinationType;
}
async function onContinueAddClicked() {
let newDestination;
switch (typeSelectValue.value) {
case MessageEventBusDestinationTypeNames.syslog:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSyslogOptions), {
id: destination.id,
});
break;
case MessageEventBusDestinationTypeNames.sentry:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSentryOptions), {
id: destination.id,
});
break;
case MessageEventBusDestinationTypeNames.webhook:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationWebhookOptions), {
id: destination.id,
});
break;
}
if (newDestination) {
headerLabel.value = newDestination?.label ?? headerLabel.value;
setupNode(newDestination);
}
}
function valueChanged(parameterData: IUpdateInformation) {
unchanged.value = false;
testMessageSent.value = false;
const newValue: NodeParameterValue = parameterData.value as string | number;
const parameterPath = parameterData.name?.startsWith('parameters.')
? parameterData.name.split('.').slice(1).join('.')
: parameterData.name || '';
const nodeParametersCopy = deepCopy(nodeParameters.value);
if (parameterData.value === undefined && parameterPath.match(/(.*)\[(\d+)\]$/)) {
const path = parameterPath.match(
/(.*)\[(\d+)\]$/,
)?.[1] as keyof MessageEventBusDestinationOptions;
const index = parseInt(parameterPath.match(/(.*)\[(\d+)\]$/)?.[2] ?? '0', 10);
const data = get(nodeParametersCopy, path);
if (Array.isArray(data)) {
data.splice(index, 1);
nodeParametersCopy[path] = data as never;
}
} else {
if (newValue === undefined) {
unset(nodeParametersCopy, parameterPath);
} else {
set(nodeParametersCopy, parameterPath, newValue);
}
}
nodeParameters.value = deepCopy(nodeParametersCopy);
workflowsStore.updateNodeProperties({
name: node.value.name,
properties: { parameters: nodeParameters.value as unknown as IDataObject, position: [0, 0] },
});
if (hasOnceBeenSaved.value) {
logStreamingStore.updateDestination(nodeParameters.value);
}
}
async function sendTestEvent() {
testMessageResult.value = await logStreamingStore.sendTestMessage(nodeParameters.value);
testMessageSent.value = true;
}
async function removeThis() {
const deleteConfirmed = await confirm(
i18n.baseText('settings.log-streaming.destinationDelete.message', {
interpolate: { destinationName: destination.label! },
}),
i18n.baseText('settings.log-streaming.destinationDelete.headline'),
{
type: 'warning',
confirmButtonText: i18n.baseText(
'settings.log-streaming.destinationDelete.confirmButtonText',
),
cancelButtonText: i18n.baseText('settings.log-streaming.destinationDelete.cancelButtonText'),
},
);
if (deleteConfirmed !== MODAL_CONFIRM) {
return;
} else {
callEventBus('remove', destination.id);
uiStore.closeModal(LOG_STREAM_MODAL_KEY);
uiStore.stateIsDirty = false;
}
}
function onModalClose() {
if (!hasOnceBeenSaved.value) {
workflowsStore.removeNode(node.value);
if (nodeParameters.value.id && typeof nodeParameters.value.id !== 'object') {
logStreamingStore.removeDestination(nodeParameters.value.id.toString());
}
}
ndvStore.activeNodeName = null;
callEventBus('closing', destination.id);
uiStore.stateIsDirty = false;
}
async function saveDestination() {
if (unchanged.value || !destination.id) {
return;
}
const saveResult = await logStreamingStore.saveDestination(nodeParameters.value);
if (saveResult) {
hasOnceBeenSaved.value = true;
testMessageSent.value = false;
unchanged.value = true;
callEventBus('destinationWasSaved', destination.id);
uiStore.stateIsDirty = false;
const destinationType = (
nodeParameters.value.__type && typeof nodeParameters.value.__type !== 'object'
? `${nodeParameters.value.__type}`
: 'unknown'
)
.replace('$$MessageEventBusDestination', '')
.toLowerCase();
const isComplete = () => {
if (isTypeWebhook.value) {
const webhookDestination = destination as MessageEventBusDestinationWebhookOptions;
return webhookDestination.url !== '';
} else if (isTypeSentry.value) {
const sentryDestination = destination as MessageEventBusDestinationSentryOptions;
return sentryDestination.dsn !== '';
} else if (isTypeSyslog.value) {
const syslogDestination = destination as MessageEventBusDestinationSyslogOptions;
return (
syslogDestination.host !== '' &&
syslogDestination.port !== undefined &&
// @ts-expect-error TODO: fix this typing
syslogDestination.protocol !== '' &&
syslogDestination.facility !== undefined &&
syslogDestination.app_name !== ''
);
}
return false;
};
telemetry.track('User updated log streaming destination', {
instance_id: useRootStore().instanceId,
destination_type: destinationType,
is_complete: isComplete(),
is_active: destination.enabled,
});
}
}
function callEventBus(event: string, data: unknown) {
if (eventBus) {
eventBus.emit(event, data);
}
}
</script>
<template>
@ -381,7 +378,7 @@ export default defineComponent({
<div :class="$style.destinationInfo">
<InlineNameEdit
:model-value="headerLabel"
:subtitle="!isTypeAbstract ? $locale.baseText(typeLabelName) : 'Select type'"
:subtitle="!isTypeAbstract ? i18n.baseText(typeLabelName) : 'Select type'"
:readonly="isTypeAbstract"
type="Credential"
data-test-id="subtitle-showing-type"
@ -406,7 +403,7 @@ export default defineComponent({
<template v-if="canManageLogStreaming">
<n8n-icon-button
v-if="nodeParameters && hasOnceBeenSaved"
:title="$locale.baseText('settings.log-streaming.delete')"
:title="i18n.baseText('settings.log-streaming.delete')"
icon="trash"
type="tertiary"
:disabled="isSaving"
@ -417,7 +414,7 @@ export default defineComponent({
<SaveButton
:saved="unchanged && hasOnceBeenSaved"
:disabled="isTypeAbstract || unchanged"
:saving-label="$locale.baseText('settings.log-streaming.saving')"
:saving-label="i18n.baseText('settings.log-streaming.saving')"
data-test-id="destination-save-button"
@click="saveDestination"
/>
@ -432,8 +429,8 @@ export default defineComponent({
<template v-if="isTypeAbstract">
<n8n-input-label
:class="$style.typeSelector"
:label="$locale.baseText('settings.log-streaming.selecttype')"
:tooltip-text="$locale.baseText('settings.log-streaming.selecttypehint')"
:label="i18n.baseText('settings.log-streaming.selecttype')"
:tooltip-text="i18n.baseText('settings.log-streaming.selecttypehint')"
:bold="false"
size="medium"
:underline="false"
@ -450,7 +447,7 @@ export default defineComponent({
v-for="option in typeSelectOptions || []"
:key="option.value"
:value="option.value"
:label="$locale.baseText(option.label)"
:label="i18n.baseText(option.label)"
/>
</n8n-select>
<div class="mt-m text-right">
@ -460,7 +457,7 @@ export default defineComponent({
:disabled="!typeSelectValue"
@click="onContinueAddClicked"
>
{{ $locale.baseText(`settings.log-streaming.continue`) }}
{{ i18n.baseText(`settings.log-streaming.continue`) }}
</n8n-button>
</div>
</n8n-input-label>
@ -505,7 +502,7 @@ export default defineComponent({
<div class="">
<n8n-input-label
class="mb-m mt-m"
:label="$locale.baseText('settings.log-streaming.tab.events.title')"
:label="i18n.baseText('settings.log-streaming.tab.events.title')"
:bold="true"
size="medium"
:underline="false"

133
packages/editor-ui/src/composables/usePinnedData.test.ts
View file

@ -3,9 +3,10 @@ import { setActivePinia, createPinia } from 'pinia';
import { ref } from 'vue';
import { usePinnedData } from '@/composables/usePinnedData';
import type { INodeUi } from '@/Interface';
import { MAX_PINNED_DATA_SIZE } from '@/constants';
import { HTTP_REQUEST_NODE_TYPE, IF_NODE_TYPE, MAX_PINNED_DATA_SIZE } from '@/constants';
import { useWorkflowsStore } from '@/stores/workflows.store';
import { useTelemetry } from '@/composables/useTelemetry';
import { NodeConnectionType, STICKY_NODE_TYPE, type INodeTypeDescription } from 'n8n-workflow';
vi.mock('@/composables/useToast', () => ({ useToast: vi.fn(() => ({ showError: vi.fn() })) }));
vi.mock('@/composables/useI18n', () => ({
@ -17,6 +18,13 @@ vi.mock('@/composables/useExternalHooks', () => ({
})),
}));
const getNodeType = vi.fn();
vi.mock('@/stores/nodeTypes.store', () => ({
useNodeTypesStore: vi.fn(() => ({
getNodeType,
})),
}));
describe('usePinnedData', () => {
beforeEach(() => {
setActivePinia(createPinia());
@ -133,4 +141,127 @@ describe('usePinnedData', () => {
expect(spy).toHaveBeenCalled();
});
});
describe('canPinData()', () => {
afterEach(() => {
vi.clearAllMocks();
});
it('allows pin on single output', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
parameters: {},
onError: 'stopWorkflow',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(true);
expect(canPinNode(false, 0)).toBe(true);
// validate out of range index
expect(canPinNode(false, 1)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('allows pin on one main and one error output', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
parameters: {},
onError: 'continueErrorOutput',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(true);
expect(canPinNode(false, 0)).toBe(true);
expect(canPinNode(false, 1)).toBe(false);
// validate out of range index
expect(canPinNode(false, 2)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('does not allow pin on two main outputs', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: IF_NODE_TYPE,
parameters: {},
onError: 'stopWorkflow',
} as INodeUi);
getNodeType.mockReturnValue(
makeNodeType([NodeConnectionType.Main, NodeConnectionType.Main], IF_NODE_TYPE),
);
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
expect(canPinNode(false, 1)).toBe(false);
// validate out of range index
expect(canPinNode(false, 2)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('does not allow pin on denylisted node', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: STICKY_NODE_TYPE,
} as INodeUi);
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
});
it('does not allow pin with checkDataEmpty and no pin', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode(true)).toBe(false);
expect(canPinNode(true, 0)).toBe(false);
});
it('does not allow pin without output', async () => {
const node = ref({
name: 'zero output node',
typeVersion: 1,
type: 'n8n-nodes-base.stopAndError',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([], 'n8n-nodes-base.stopAndError'));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
expect(canPinNode(false, 1)).toBe(false);
});
});
});
const makeNodeType = (outputs: NodeConnectionType[], name: string) =>
({
displayName: name,
name,
version: [1],
inputs: [],
outputs,
properties: [],
defaults: { color: '', name: '' },
group: [],
description: '',
}) as INodeTypeDescription;

27
packages/editor-ui/src/composables/usePinnedData.ts
View file

@ -75,9 +75,9 @@ export function usePinnedData(
);
});
function canPinNode(checkDataEmpty = false) {
function canPinNode(checkDataEmpty = false, outputIndex?: number) {
const targetNode = unref(node);
if (targetNode === null) return false;
if (targetNode === null || PIN_DATA_NODE_TYPES_DENYLIST.includes(targetNode.type)) return false;
const nodeType = useNodeTypesStore().getNodeType(targetNode.type, targetNode.typeVersion);
const dataToPin = getInputDataWithPinned(targetNode);
@ -85,14 +85,25 @@ export function usePinnedData(
if (!nodeType || (checkDataEmpty && dataToPin.length === 0)) return false;
const workflow = workflowsStore.getCurrentWorkflow();
const outputs = NodeHelpers.getNodeOutputs(workflow, targetNode, nodeType);
const mainOutputs = outputs.filter((output) =>
typeof output === 'string'
? output === NodeConnectionType.Main
: output.type === NodeConnectionType.Main,
const outputs = NodeHelpers.getNodeOutputs(workflow, targetNode, nodeType).map((output) =>
typeof output === 'string' ? { type: output } : output,
);
return mainOutputs.length === 1 && !PIN_DATA_NODE_TYPES_DENYLIST.includes(targetNode.type);
const mainOutputs = outputs.filter(
(output) => output.type === NodeConnectionType.Main && output.category !== 'error',
);
let indexAcceptable = true;
if (outputIndex !== undefined) {
const output = outputs[outputIndex];
if (outputs[outputIndex] === undefined) return false;
indexAcceptable = output.type === NodeConnectionType.Main && output.category !== 'error';
}
return mainOutputs.length === 1 && indexAcceptable;
}
function isValidJSON(data: string): boolean {

2
packages/editor-ui/src/plugins/i18n/locales/en.json
View file

@ -1498,7 +1498,7 @@
"pushConnection.executionFailed": "Execution failed",
"pushConnection.executionFailed.message": "There might not be enough memory to finish the execution. Tips for avoiding this <a target=\"_blank\" href=\"https://docs.n8n.io/flow-logic/error-handling/memory-errors/\">here</a>",
"pushConnection.executionError": "There was a problem executing the workflow{error}",
"pushConnection.executionError.openNode": " <a data-action='openNodeDetail' data-action-parameter-node='{node}'>Open node</a>",
"pushConnection.executionError.openNode": "Open errored node",
"pushConnection.executionError.details": "<br /><strong>{details}</strong>",
"prompts.productTeamMessage": "Our product team will get in touch personally",
"prompts.npsSurvey.recommendationQuestion": "How likely are you to recommend n8n to a friend or colleague?",

2
packages/node-dev/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-node-dev",
"version": "1.66.0",
"version": "1.67.0",
"description": "CLI to simplify n8n credentials/node development",
"main": "dist/src/index",
"types": "dist/src/index.d.ts",

2
packages/nodes-base/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-nodes-base",
"version": "1.66.0",
"version": "1.67.0",
"description": "Base nodes of n8n",
"main": "index.js",
"scripts": {

2
packages/workflow/package.json
View file

@ -1,6 +1,6 @@
{
"name": "n8n-workflow",
"version": "1.65.0",
"version": "1.66.0",
"description": "Workflow base code of n8n",
"main": "dist/index.js",
"module": "src/index.ts",