perf(core): Deduplicate task runner data request response (no-changelog) (#11583)

Co-authored-by: Iván Ovejero <ivov.src@gmail.com>
2025-12-17 01:56:46 +00:00 · 2024-11-07 11:24:00 +02:00
parent 8cba100488
commit 3111dece72
15 changed files with 672 additions and 572 deletions
--- a/packages/@n8n/task-runner/src/data-request/data-request-response-reconstruct.ts
+++ b/packages/@n8n/task-runner/src/data-request/data-request-response-reconstruct.ts
@@ -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,
+		};
+	}
+}
--- a/packages/@n8n/task-runner/src/index.ts
+++ b/packages/@n8n/task-runner/src/index.ts
@@ -1,3 +1,4 @@
 export * from './task-runner';
 export * from './runner-types';
 export * from './message-types';
+export * from './data-request/data-request-response-reconstruct';
--- a/packages/@n8n/task-runner/src/js-task-runner/tests/js-task-runner.test.ts
+++ b/packages/@n8n/task-runner/src/js-task-runner/tests/js-task-runner.test.ts
@@ -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);

--- a/packages/@n8n/task-runner/src/js-task-runner/tests/test-data.ts
+++ b/packages/@n8n/task-runner/src/js-task-runner/tests/test-data.ts
@@ -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,
 	};
--- a/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/tests/built-ins-parser.test.ts
+++ b/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/tests/built-ins-parser.test.ts
@@ -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,
--- a/packages/@n8n/task-runner/src/js-task-runner/js-task-runner.ts
+++ b/packages/@n8n/task-runner/src/js-task-runner/js-task-runner.ts
@@ -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);
+		}
+	}
 }
--- a/packages/@n8n/task-runner/src/runner-types.ts
+++ b/packages/@n8n/task-runner/src/runner-types.ts
@@ -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;