From 894709233c0a57d8788feee5b2a31f71e0f86355 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Ovejero?= Date: Mon, 3 Aug 2026 12:30:45 +0200 Subject: [PATCH] feat(core): Resolve expressions in v1 step executor (no-changelog) (#35338) --- .../src/__tests__/fixtures.ts | 27 ++- .../src/__tests__/setup-vm-evaluator.ts | 16 ++ .../src/__tests__/v1-adapters.test.ts | 105 +++++++++ .../src/__tests__/v1-step-executor.test.ts | 151 +++++++++--- .../src/constants.ts | 3 + .../node-engine-compatibility/src/errors.ts | 16 +- .../node-engine-compatibility/src/index.ts | 2 +- .../node-engine-compatibility/src/types.ts | 31 ++- .../src/v1-adapters.ts | 221 ++++++++++++++++++ .../src/v1-step-executor.ts | 143 ++++-------- .../src/v1-workflow-converter.ts | 5 +- .../vitest.config.ts | 1 + packages/workflow/src/expression.ts | 15 ++ packages/workflow/src/workflow-expression.ts | 4 + .../test/expression-with-isolate.test.ts | 55 +++++ 15 files changed, 648 insertions(+), 147 deletions(-) create mode 100644 packages/@n8n/node-engine-compatibility/src/__tests__/setup-vm-evaluator.ts create mode 100644 packages/@n8n/node-engine-compatibility/src/__tests__/v1-adapters.test.ts create mode 100644 packages/@n8n/node-engine-compatibility/src/constants.ts create mode 100644 packages/@n8n/node-engine-compatibility/src/v1-adapters.ts create mode 100644 packages/workflow/test/expression-with-isolate.test.ts diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts index e02477a2688..cc5e9291f0a 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts @@ -1,5 +1,6 @@ import type { JsonObject, JsonValue, StepExecutionRequest, WorkflowGraph } from '@n8n/engine'; import type { ExecuteContext } from 'n8n-core'; +import { UnrecognizedNodeTypeError } from 'n8n-core'; import { NoOp } from 'n8n-nodes-base/nodes/NoOp/NoOp.node'; import type { CloseFunction, @@ -16,6 +17,8 @@ import type { } from 'n8n-workflow'; import { Node, NodeConnectionTypes } from 'n8n-workflow'; +import { V1StepExecutor } from '../v1-step-executor'; + class EchoParam implements INodeType { description = { displayName: 'Echo Param', @@ -193,8 +196,16 @@ const registry = new Map([ ]); export const testNodeTypes: INodeTypes = { - getByName: (type: string): INodeType | IVersionedNodeType => registry.get(type)!, - getByNameAndVersion: (type: string): INodeType => registry.get(type)!, + getByName: (type: string): INodeType | IVersionedNodeType => { + const nodeType = registry.get(type); + if (!nodeType) throw new UnrecognizedNodeTypeError('test', type); + return nodeType; + }, + getByNameAndVersion: (type: string): INodeType => { + const nodeType = registry.get(type); + if (!nodeType) throw new UnrecognizedNodeTypeError('test', type); + return nodeType; + }, getKnownTypes: (): IDataObject => ({}), }; @@ -239,7 +250,17 @@ export const stepRequest = ( ): StepExecutionRequest => ({ node: graph.nodes.find((n) => n.id === nodeId)!, inputs, - context: { executionId: 'exec-1', stepId: 'step-1', workflowId: 'wf-1', mode: 'manual' }, + context: { executionId: 'exec-1', stepId: nodeId, workflowId: 'wf-1', mode: 'manual' }, }); +export const testStepExecutor = ( + graph: WorkflowGraph, + outputsByStepId: Record = {}, +): V1StepExecutor => + new V1StepExecutor({ + nodeTypes: testNodeTypes, + additionalDataFactory: testAdditionalDataFactory, + loadStepData: async () => await Promise.resolve({ graph, outputsByStepId }), + }); + export const items = (...objects: JsonObject[]): JsonValue => [objects.map((json) => ({ json }))]; diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/setup-vm-evaluator.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/setup-vm-evaluator.ts new file mode 100644 index 00000000000..48f8e9e75a5 --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/setup-vm-evaluator.ts @@ -0,0 +1,16 @@ +import { Expression } from 'n8n-workflow'; +import { afterAll, beforeAll } from 'vitest'; + +beforeAll(async () => { + await Expression.initExpressionEngine({ + engine: 'vm', + poolSize: 1, + maxCodeCacheSize: 1024, + bridgeTimeout: 5000, + bridgeMemoryLimit: 128, + }); +}); + +afterAll(async () => { + await Expression.disposeExpressionEngine(); +}); diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-adapters.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-adapters.test.ts new file mode 100644 index 00000000000..b392ac72349 --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-adapters.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it } from 'vitest'; + +import { toV1Execution, toV1Sources } from '../v1-adapters'; +import { V1WorkflowConverter } from '../v1-workflow-converter'; +import { items, v1Workflow } from './fixtures'; + +const converter = new V1WorkflowConverter(); + +// The shared fixture graph, in the notation of the converter tests +// (oN / iN mark output / input slots, (back) the loop-return edge): +// +// ┌───────┐ ┌─┐ o1 i1 ┌─┐ ┌────┐ o1 ┌────┐ +// │Trigger├───►│A├─────────►│B├───►│Loop├──────►│Body│ +// └───────┘ └─┘ └─┘ └─▲──┘ └──┬─┘ +// └───(back)────┘ +const graph = converter.convert( + v1Workflow( + [ + { id: 't', name: 'Trigger', type: 'n8n-nodes-base.manualTrigger' }, + { id: 'a', name: 'A', type: 'test.echoParam' }, + { id: 'b', name: 'B', type: 'test.echoParam' }, + { id: 'loop', name: 'Loop', type: 'n8n-nodes-base.splitInBatches', typeVersion: 3 }, + { id: 'body', name: 'Body', type: 'n8n-nodes-base.noOp' }, + ], + { + Trigger: { main: [[{ node: 'A', type: 'main', index: 0 }]] }, + A: { main: [[], [{ node: 'B', type: 'main', index: 1 }]] }, + B: { main: [[{ node: 'Loop', type: 'main', index: 0 }]] }, + Loop: { main: [[], [{ node: 'Body', type: 'main', index: 0 }]] }, + Body: { main: [[{ node: 'Loop', type: 'main', index: 0 }]] }, + }, + ), +); + +describe('toV1Execution', () => { + it('rebuilds v1 nodes, including a stub for the trigger', () => { + const execution = toV1Execution(graph, {}); + + expect(execution.nodes.map((node) => node.name).sort()).toEqual([ + 'A', + 'B', + 'Body', + 'Loop', + 'Trigger', + ]); + expect(execution.nodes.find((node) => node.name === 'Trigger')).toMatchObject({ + type: 'n8n-nodes-base.manualTrigger', + }); + expect(execution.nodes.find((node) => node.name === 'A')).toMatchObject({ + type: 'test.echoParam', + typeVersion: 1, + continueOnFail: false, + }); + }); + + it('rebuilds name-keyed connections preserving slots', () => { + const execution = toV1Execution(graph, {}); + + expect(execution.connections.A).toEqual({ + main: [[], [{ node: 'B', type: 'main', index: 1 }]], + }); + }); + + it('rebuilds run data with input provenance from the edges', () => { + const execution = toV1Execution(graph, { a: items({ x: 1 }) }); + + expect(execution.runData).toEqual({ + A: [ + { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [{ previousNode: 'Trigger', previousNodeOutput: 0 }], + data: { main: [[{ json: { x: 1 } }]] }, + }, + ], + }); + expect(toV1Sources(graph).get('b')).toEqual([ + null, + { previousNode: 'A', previousNodeOutput: 1 }, + ]); + }); + + it('keeps back edges in connections but not in provenance', () => { + const execution = toV1Execution(graph, {}); + + expect(execution.connections.Body).toEqual({ + main: [[{ node: 'Loop', type: 'main', index: 0 }]], + }); + expect(toV1Sources(graph).get('loop')).toEqual([{ previousNode: 'B', previousNodeOutput: 0 }]); + }); + + it('omits nodes whose config is unreadable', () => { + const brokenGraph = { + nodes: [ + { id: 'ok', name: 'OK', type: 'v1-node' as const, config: graph.nodes[1].config }, + { id: 'bad', name: 'Bad', type: 'v1-node' as const, config: 'nonsense' }, + ], + edges: [], + }; + + const execution = toV1Execution(brokenGraph, {}); + expect(execution.nodes.map((node) => node.name)).toEqual(['OK']); + }); +}); diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-step-executor.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-step-executor.test.ts index b8bdca2e1a2..c405d1a2dc7 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-step-executor.test.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-step-executor.test.ts @@ -1,28 +1,20 @@ import type { WorkflowGraph } from '@n8n/engine'; -import { describe, expect, it } from 'vitest'; +import { UnrecognizedNodeTypeError } from 'n8n-core'; +import type { IDataObject } from 'n8n-workflow'; +import { Expression, ExpressionError } from 'n8n-workflow'; +import { describe, expect, it, vi } from 'vitest'; import { EngineRequestNotSupportedError, MalformedStepConfigError, - UnknownNodeTypeError, UnsupportedNodeTypeError, UnsupportedStepTypeError, + VmExpressionEngineRequiredError, } from '../errors'; -import { V1StepExecutor } from '../v1-step-executor'; import { V1WorkflowConverter } from '../v1-workflow-converter'; -import { - items, - stepRequest, - testAdditionalDataFactory, - testNodeTypes, - v1Workflow, -} from './fixtures'; +import { items, stepRequest, testStepExecutor, v1Workflow } from './fixtures'; const converter = new V1WorkflowConverter(); -const executor = new V1StepExecutor({ - nodeTypes: testNodeTypes, - additionalDataFactory: testAdditionalDataFactory, -}); const graphWith = (type: string, parameters = {}): WorkflowGraph => converter.convert( @@ -33,9 +25,22 @@ const graphWith = (type: string, parameters = {}): WorkflowGraph => ); describe('V1StepExecutor', () => { + it('rejects legacy expression engine', async () => { + vi.spyOn(Expression, 'getActiveImplementation').mockReturnValue('legacy'); + try { + const graph = graphWith('test.echoParam', { message: 'hi' }); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({}))); + await expect(execution).rejects.toThrow(VmExpressionEngineRequiredError); + } finally { + vi.restoreAllMocks(); + } + }); + it('resolves `getNodeParameter` per item', async () => { const graph = graphWith('test.echoParam', { message: 'hi' }); - const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 }, { a: 2 }))); + const result = await testStepExecutor(graph).execute( + stepRequest(graph, 'n', items({ a: 1 }, { a: 2 })), + ); expect(result.outputs).toEqual([[{ json: { message: 'hi' } }, { json: { message: 'hi' } }]]); // eslint-disable-next-line n8n-local-rules/no-json-parse-json-stringify expect(JSON.parse(JSON.stringify(result.outputs))).toEqual(result.outputs); @@ -43,14 +48,14 @@ describe('V1StepExecutor', () => { it('guarantees one item for empty input', async () => { const graph = graphWith('test.echoParam', { message: 'solo' }); - const result = await executor.execute(stepRequest(graph, 'n', [])); + const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', [])); expect(result.outputs).toEqual([[{ json: { message: 'solo' } }]]); }); it('rejects non-v1-node steps', async () => { const graph = graphWith('n8n-nodes-base.noOp'); const triggerStep = stepRequest(graph, 't', []); - const execution = executor.execute(triggerStep); + const execution = testStepExecutor(graph).execute(triggerStep); await expect(execution).rejects.toThrow(UnsupportedStepTypeError); }); @@ -67,31 +72,31 @@ describe('V1StepExecutor', () => { const graph = graphWith('n8n-nodes-base.noOp'); const request = stepRequest(graph, 'n', []); request.node = { ...request.node, config }; - const execution = executor.execute(request); + const execution = testStepExecutor(graph).execute(request); await expect(execution).rejects.toThrow(MalformedStepConfigError); }); it('rejects unknown node types', async () => { const graph = graphWith('test.doesNotExist'); - const execution = executor.execute(stepRequest(graph, 'n', [])); - await expect(execution).rejects.toThrow(UnknownNodeTypeError); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', [])); + await expect(execution).rejects.toThrow(UnrecognizedNodeTypeError); }); it('rejects node types without an execute method', async () => { const graph = graphWith('test.noExecute'); - const execution = executor.execute(stepRequest(graph, 'n', [])); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', [])); await expect(execution).rejects.toThrow(UnsupportedNodeTypeError); }); it('propagates node errors per the IStepExecutor failure contract', async () => { const graph = graphWith('test.alwaysFails'); - const execution = executor.execute(stepRequest(graph, 'n', [])); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', [])); await expect(execution).rejects.toThrow('boom from node'); }); it('invokes new-style Node subclasses with the context as argument', async () => { const graph = graphWith('test.newStyleEcho'); - const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); expect(result.outputs).toEqual([[{ json: { a: 1, newStyle: true } }]]); }); @@ -103,25 +108,27 @@ describe('V1StepExecutor', () => { (workflow.nodes[1] as { continueOnFail?: boolean }).continueOnFail = true; const graph = converter.convert(workflow); - const result = await executor.execute(stepRequest(graph, 'n', items({ keep: 'me' }))); + const result = await testStepExecutor(graph).execute( + stepRequest(graph, 'n', items({ keep: 'me' })), + ); expect(result.outputs).toEqual([[{ json: { keep: 'me' } }]]); }); it('propagates cleanup errors when the node succeeded', async () => { const graph = graphWith('test.succeedsWithFailingCleanup'); - const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); await expect(execution).rejects.toThrow('cleanup boom'); }); it('preserves the node error over cleanup errors when both fail', async () => { const graph = graphWith('test.failsWithFailingCleanup'); - const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); await expect(execution).rejects.toThrow('boom from node'); }); it('throws on EngineRequest results instead of dropping them', async () => { const graph = graphWith('test.returnsEngineRequest'); - const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); await expect(execution).rejects.toThrow(EngineRequestNotSupportedError); }); @@ -133,7 +140,7 @@ describe('V1StepExecutor', () => { (workflow.nodes[1] as { continueOnFail?: boolean }).continueOnFail = true; const graph = converter.convert(workflow); - const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); await expect(execution).rejects.toThrow(EngineRequestNotSupportedError); }); @@ -158,7 +165,7 @@ describe('V1StepExecutor', () => { ), ); - const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 }))); + const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 }))); expect(result.outputs).toEqual([[{ json: { a: 1 } }], [{ json: { a: 1, second: true } }]]); }); @@ -178,10 +185,90 @@ describe('V1StepExecutor', () => { ), ); - const execution = executor.execute(stepRequest(graph, 'loop', items({ a: 1 }))); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'loop', items({ a: 1 }))); await expect(execution).rejects.toThrow(UnsupportedStepTypeError); }); + describe('expressions', () => { + const expressionWorkflow = (bParameters: IDataObject) => + converter.convert( + v1Workflow( + [ + { id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' }, + { id: 'a', name: 'A', type: 'test.echoParam', parameters: { message: 'from-A' } }, + { id: 'b', name: 'B', type: 'test.echoParam', parameters: bParameters }, + ], + { + Manual: { main: [[{ node: 'A', type: 'main', index: 0 }]] }, + A: { main: [[{ node: 'B', type: 'main', index: 0 }]] }, + }, + ), + ); + + it('resolves $json per item against the step inputs', async () => { + const graph = graphWith('test.echoParam', { message: '={{ $json.a }}' }); + const result = await testStepExecutor(graph).execute( + stepRequest(graph, 'n', items({ a: 1 }, { a: 2 })), + ); + expect(result.outputs).toEqual([[{ json: { message: 1 } }, { json: { message: 2 } }]]); + }); + + it('resolves $(...) against a completed step output, convert to execute', async () => { + const graph = expressionWorkflow({ message: "={{ $('A').first().json.message }}" }); + + const aResult = await testStepExecutor(graph).execute(stepRequest(graph, 'a', items({}))); + const bResult = await testStepExecutor(graph, { a: aResult.outputs }).execute( + stepRequest(graph, 'b', aResult.outputs), + ); + + expect(bResult.outputs).toEqual([[{ json: { message: 'from-A' } }]]); + }); + + it('resolves $prevNode from the input provenance in the graph', async () => { + const graph = expressionWorkflow({ message: '={{ $prevNode.name }}' }); + const result = await testStepExecutor(graph).execute(stepRequest(graph, 'b', items({}))); + expect(result.outputs).toEqual([[{ json: { message: 'A' } }]]); + }); + + it('rejects a reference to an unexecuted node', async () => { + const graph = expressionWorkflow({ message: "={{ $('A').first().json.message }}" }); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'b', items({}))); + await expect(execution).rejects.toThrow("Node 'A' hasn't been executed"); + }); + + it('rejects a reference to a node absent from the graph', async () => { + const graph = graphWith('test.echoParam', { message: "={{ $('Ghost').first().json.x }}" }); + const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({}))); + await expect(execution).rejects.toThrow(ExpressionError); + }); + + it('resolves a sibling reference even when its node type is not installed', async () => { + const graph = converter.convert( + v1Workflow( + [ + { id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' }, + { id: 'a', name: 'A', type: 'community.notInstalled' }, + { + id: 'b', + name: 'B', + type: 'test.echoParam', + parameters: { message: "={{ $('A').first().json.tag }}" }, + }, + ], + { + Manual: { main: [[{ node: 'A', type: 'main', index: 0 }]] }, + A: { main: [[{ node: 'B', type: 'main', index: 0 }]] }, + }, + ), + ); + + const result = await testStepExecutor(graph, { a: items({ tag: 'ran-before' }) }).execute( + stepRequest(graph, 'b', items({})), + ); + expect(result.outputs).toEqual([[{ json: { message: 'ran-before' } }]]); + }); + }); + it('honors onError=continueRegularOutput as passthrough', async () => { const workflow = v1Workflow([ { id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' }, @@ -190,7 +277,9 @@ describe('V1StepExecutor', () => { (workflow.nodes[1] as { onError?: string }).onError = 'continueRegularOutput'; const graph = converter.convert(workflow); - const result = await executor.execute(stepRequest(graph, 'n', items({ keep: 'me' }))); + const result = await testStepExecutor(graph).execute( + stepRequest(graph, 'n', items({ keep: 'me' })), + ); expect(result.outputs).toEqual([[{ json: { keep: 'me' } }]]); }); }); diff --git a/packages/@n8n/node-engine-compatibility/src/constants.ts b/packages/@n8n/node-engine-compatibility/src/constants.ts new file mode 100644 index 00000000000..4e785508b3c --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/constants.ts @@ -0,0 +1,3 @@ +export const MANUAL_TRIGGER_TYPE = 'n8n-nodes-base.manualTrigger'; +export const SPLIT_IN_BATCHES_TYPE = 'n8n-nodes-base.splitInBatches'; +export const MAIN_CONNECTION_TYPE = 'main'; diff --git a/packages/@n8n/node-engine-compatibility/src/errors.ts b/packages/@n8n/node-engine-compatibility/src/errors.ts index 16c78854558..dbf38e25db3 100644 --- a/packages/@n8n/node-engine-compatibility/src/errors.ts +++ b/packages/@n8n/node-engine-compatibility/src/errors.ts @@ -1,4 +1,4 @@ -import { UserError } from 'n8n-workflow'; +import { UnexpectedError, UserError } from 'n8n-workflow'; const quote = (names: string[]) => names.map((name) => `"${name}"`).join(', '); @@ -52,18 +52,20 @@ export class MalformedStepConfigError extends UserError { } } -export class UnknownNodeTypeError extends UserError { - constructor(nodeType: string) { - super(`Unknown node type "${nodeType}"`); - } -} - export class UnsupportedNodeTypeError extends UserError { constructor(nodeType: string) { super(`Node type "${nodeType}" has no execute method and cannot run as a step`); } } +export class VmExpressionEngineRequiredError extends UnexpectedError { + constructor() { + super( + 'V1StepExecutor requires the VM expression engine. Initialize it via `Expression.initExpressionEngine()` before executing steps.', + ); + } +} + export class EngineRequestNotSupportedError extends UserError { constructor(nodeType: string) { super( diff --git a/packages/@n8n/node-engine-compatibility/src/index.ts b/packages/@n8n/node-engine-compatibility/src/index.ts index ce9df6da461..43109cd2a75 100644 --- a/packages/@n8n/node-engine-compatibility/src/index.ts +++ b/packages/@n8n/node-engine-compatibility/src/index.ts @@ -1,4 +1,4 @@ export { V1WorkflowConverter } from './v1-workflow-converter'; export { V1StepExecutor } from './v1-step-executor'; export { UnsupportedTriggerError, UnsupportedWorkflowError } from './errors'; -export type { V1StepExecutorDeps } from './types'; +export type { StepData, StepDataLoader, V1StepExecutorDeps } from './types'; diff --git a/packages/@n8n/node-engine-compatibility/src/types.ts b/packages/@n8n/node-engine-compatibility/src/types.ts index aa5bc971846..480073cfa07 100644 --- a/packages/@n8n/node-engine-compatibility/src/types.ts +++ b/packages/@n8n/node-engine-compatibility/src/types.ts @@ -1,11 +1,16 @@ -import type { StepExecutionContext } from '@n8n/engine'; +import type { JsonValue, StepExecutionContext, WorkflowGraph } from '@n8n/engine'; import type { ExecuteContext } from 'n8n-core'; import type { + INode, INodeExecutionData, INodeParameters, INodeType, INodeTypes, + IRunData, + ISourceData, + IWorkflowBase, IWorkflowExecuteAdditionalData, + Workflow, } from 'n8n-workflow'; export interface V1NodeStepConfig { @@ -15,24 +20,38 @@ export interface V1NodeStepConfig { continueOnFail: boolean; } +export interface V1Execution extends Pick { + runData: IRunData; +} + +export interface StepData { + graph: WorkflowGraph; + outputsByStepId: Record; +} + +export type StepDataLoader = (context: StepExecutionContext) => Promise; + export interface V1StepExecutorDeps { nodeTypes: INodeTypes; additionalDataFactory: (executionId: string) => Promise; + loadStepData: StepDataLoader; } export type ExecutableNodeType = INodeType & { execute: NonNullable }; export type NodeRunResult = { ok: true; value: unknown } | { ok: false; error: unknown }; -export interface CreateNodeExecuteContextParams { - nodeId: string; - nodeName: string; - stepConfig: V1NodeStepConfig; +export interface CreateExecuteContextParams { + node: INode; + workflow: Workflow; + execution: V1Execution; + source: Array; + additionalData: IWorkflowExecuteAdditionalData; itemsByConnection: INodeExecutionData[][]; stepContext: StepExecutionContext; } export interface RunNodeParams { nodeType: ExecutableNodeType; - nodeExecuteContext: ExecuteContext; + context: ExecuteContext; } diff --git a/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts b/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts new file mode 100644 index 00000000000..5e5e173557e --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts @@ -0,0 +1,221 @@ +/** + * Pure functions to turn engine v2 artifacts (graph, step outputs, step context) + * back into engine v1 artifacts (`INode`, `Workflow`, run data, node execute context). + */ + +import type { GraphNode, JsonValue, WorkflowGraph } from '@n8n/engine'; +import { ExecuteContext, UnrecognizedNodeTypeError } from 'n8n-core'; +import type { + IConnections, + IExecuteData, + INode, + INodeType, + INodeTypes, + IRunData, + ISourceData, + ITaskDataConnections, + WorkflowExecuteMode, +} from 'n8n-workflow'; +import { createRunExecutionData, Workflow } from 'n8n-workflow'; + +import { MAIN_CONNECTION_TYPE, MANUAL_TRIGGER_TYPE } from './constants'; +import { isV1NodeStepConfig } from './guards'; +import { fromStepInputs } from './io'; +import type { CreateExecuteContextParams, V1Execution, V1NodeStepConfig } from './types'; + +export function toV1Execution( + graph: WorkflowGraph, + outputsByStepId: Record, +): V1Execution { + return { + nodes: toV1Nodes(graph), + connections: toV1Connections(graph), + runData: toV1RunData(graph, outputsByStepId), + }; +} + +function toV1Nodes(graph: WorkflowGraph): INode[] { + return graph.nodes.flatMap((graphNode): INode[] => { + if (graphNode.type === 'trigger') { + return [ + { + id: graphNode.id, + name: graphNode.name, + type: MANUAL_TRIGGER_TYPE, + typeVersion: 1, + position: [0, 0], + parameters: {}, + }, + ]; + } + + if (!isV1NodeStepConfig(graphNode.config)) return []; + + return [toV1Node(graphNode, graphNode.config)]; + }); +} + +function toV1Connections(graph: WorkflowGraph): IConnections { + const namesById = new Map(graph.nodes.map((node) => [node.id, node.name])); + + const connections: IConnections = {}; + for (const edge of graph.edges) { + const fromName = namesById.get(edge.from); + const toName = namesById.get(edge.to); + if (fromName === undefined || toName === undefined) continue; + + const outputs = (connections[fromName] ??= { [MAIN_CONNECTION_TYPE]: [] })[ + MAIN_CONNECTION_TYPE + ]; + const outputIndex = edge.outputIndex ?? 0; + while (outputs.length <= outputIndex) outputs.push([]); + outputs[outputIndex]!.push({ + node: toName, + type: MAIN_CONNECTION_TYPE, + index: edge.inputIndex ?? 0, + }); + } + + return connections; +} + +function toV1RunData(graph: WorkflowGraph, outputsByStepId: Record): IRunData { + const namesById = new Map(graph.nodes.map((node) => [node.id, node.name])); + const sourcesByStepId = toV1Sources(graph); + + const runData: IRunData = {}; + for (const [completedStepId, outputs] of Object.entries(outputsByStepId)) { + const nodeName = namesById.get(completedStepId); + if (nodeName === undefined) continue; + + runData[nodeName] = [ + { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: sourcesByStepId.get(completedStepId) ?? [], + data: { [MAIN_CONNECTION_TYPE]: fromStepInputs(outputs) }, + }, + ]; + } + + return runData; +} + +export function toV1Sources(graph: WorkflowGraph): Map> { + const namesById = new Map(graph.nodes.map((node) => [node.id, node.name])); + + const sourcesByStepId = new Map>(); + for (const edge of graph.edges) { + if (edge.isBackEdge === true) continue; + const fromName = namesById.get(edge.from); + if (fromName === undefined) continue; + + const source = sourcesByStepId.get(edge.to) ?? []; + const inputIndex = edge.inputIndex ?? 0; + while (source.length <= inputIndex) source.push(null); + source[inputIndex] ??= { previousNode: fromName, previousNodeOutput: edge.outputIndex }; + sourcesByStepId.set(edge.to, source); + } + return sourcesByStepId; +} + +export function toV1Workflow( + workflowId: string, + node: INode, + execution: V1Execution, + nodeTypes: INodeTypes, +): Workflow { + const siblingNodes = execution.nodes.filter((executionNode) => executionNode.id !== node.id); + + return new Workflow({ + id: workflowId, + nodes: [node, ...siblingNodes], + connections: execution.connections, + active: false, + nodeTypes: tolerantNodeTypes(nodeTypes), + }); +} + +export function toV1Node(graphNode: GraphNode, config: V1NodeStepConfig): INode { + return { + id: graphNode.id, + name: graphNode.name, + type: config.nodeType, + typeVersion: config.typeVersion, + position: [0, 0], + parameters: config.parameters, + continueOnFail: config.continueOnFail, + }; +} + +/** + * In v1, an uninstalled node type _fails the whole execution upfront_, at + * `Workflow` construction. In v2, where each step builds its own `Workflow`, + * preserving that failure would mean failing at every step of the execution. + * Instead, for fault isolation, v2 maps this to `undefined`, which `Workflow` + * tolerates by design (node kept, name and data visible to expressions, only + * parameter defaults skipped), so only the missing node type step fails. + */ +function tolerantNodeTypes(nodeTypes: INodeTypes): INodeTypes { + return { + getByName: (type) => nodeTypes.getByName(type), + getKnownTypes: () => nodeTypes.getKnownTypes(), + getByNameAndVersion: (type, version) => { + try { + return nodeTypes.getByNameAndVersion(type, version); + } catch (error) { + if (error instanceof UnrecognizedNodeTypeError) return undefined as unknown as INodeType; + throw error; + } + }, + }; +} + +export function toV1ExecuteContext({ + node, + workflow, + execution, + source, + additionalData, + stepContext, + itemsByConnection, +}: CreateExecuteContextParams): ExecuteContext { + const firstInput = itemsByConnection[0] ?? []; + const connectionInputData = firstInput.length > 0 ? firstInput : [{ json: {} }]; + const inputData: ITaskDataConnections = { + main: [connectionInputData, ...itemsByConnection.slice(1)], + }; + + const runExecutionData = createRunExecutionData({ + startData: {}, + resultData: { runData: execution.runData }, + executionData: { + contextData: {}, + nodeExecutionStack: [], + metadata: {}, + waitingExecution: {}, + waitingExecutionSource: null, + }, + }); + + const executeData: IExecuteData = { + node, + data: inputData, + source: source.length > 0 ? { main: source } : null, + }; + const mode: WorkflowExecuteMode = stepContext.mode === 'production' ? 'trigger' : 'manual'; + + return new ExecuteContext( + workflow, + node, + additionalData, + mode, + runExecutionData, + 0, + connectionInputData, + inputData, + executeData, + [], + ); +} diff --git a/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts b/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts index a1c66477651..4bef4d69e03 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts @@ -4,38 +4,33 @@ import type { StepExecutionRequest, StepExecutionResult, } from '@n8n/engine'; -import { ExecuteContext } from 'n8n-core'; -import type { - IExecuteData, - INode, - INodeExecutionData, - ITaskDataConnections, - WorkflowExecuteMode, -} from 'n8n-workflow'; -import { - createRunExecutionData, - isNodeClassInstance, - UnexpectedError, - Workflow, -} from 'n8n-workflow'; +import { UnrecognizedNodeTypeError } from 'n8n-core'; +import type { INodeExecutionData } from 'n8n-workflow'; +import { Expression, isNodeClassInstance, UnexpectedError } from 'n8n-workflow'; import { EngineRequestNotSupportedError, MalformedStepConfigError, UnsupportedNodeTypeError, - UnknownNodeTypeError, UnsupportedStepTypeError, + VmExpressionEngineRequiredError, } from './errors'; import { isV1NodeStepConfig } from './guards'; import { fromStepInputs, toStepOutputs } from './io'; import type { - CreateNodeExecuteContextParams, ExecutableNodeType, NodeRunResult, RunNodeParams, V1NodeStepConfig, V1StepExecutorDeps, } from './types'; +import { + toV1ExecuteContext, + toV1Execution, + toV1Node, + toV1Sources, + toV1Workflow, +} from './v1-adapters'; /** * Runs `v1-node` steps by adapting them to the v1 node runtime. @@ -48,20 +43,37 @@ export class V1StepExecutor implements IStepExecutor { constructor(private readonly deps: V1StepExecutorDeps) {} async execute(request: StepExecutionRequest): Promise { + this.validateExpressionEngine(); const stepConfig = this.validateStepConfig(request.node); const nodeType = this.resolveNodeType(stepConfig); + const stepData = await this.deps.loadStepData(request.context); - const nodeExecuteContext = await this.createNodeExecuteContext({ - nodeId: request.node.id, - nodeName: request.node.name, + const node = toV1Node(request.node, stepConfig); + const execution = toV1Execution(stepData.graph, stepData.outputsByStepId); + const workflow = toV1Workflow(request.context.workflowId, node, execution, this.deps.nodeTypes); + + const additionalData = await this.deps.additionalDataFactory(request.context.executionId); + + const context = toV1ExecuteContext({ + node, + workflow, + execution, + source: toV1Sources(stepData.graph).get(node.id) ?? [], + additionalData, stepContext: request.context, - stepConfig, itemsByConnection: fromStepInputs(request.inputs), }); - const nodeResult = await this.runNode({ nodeType, nodeExecuteContext }); + return await workflow.expression.withIsolate(async () => { + const nodeResult = await this.runNode({ nodeType, context }); + return { outputs: toStepOutputs(nodeResult) }; + }); + } - return { outputs: toStepOutputs(nodeResult) }; + private validateExpressionEngine(): void { + if (Expression.getActiveImplementation() !== 'vm') { + throw new VmExpressionEngineRequiredError(); + } } private validateStepConfig({ type, name, config }: GraphNode): V1NodeStepConfig { @@ -76,73 +88,17 @@ export class V1StepExecutor implements IStepExecutor { typeVersion, }: V1NodeStepConfig): ExecutableNodeType { const nodeType = this.deps.nodeTypes.getByNameAndVersion(typeName, typeVersion); - if (!nodeType) throw new UnknownNodeTypeError(typeName); + + if (!nodeType) { + const [packageName, name = typeName] = typeName.split('.'); + throw new UnrecognizedNodeTypeError(packageName, name); + } + if (typeof nodeType.execute !== 'function') throw new UnsupportedNodeTypeError(typeName); return nodeType as ExecutableNodeType; } - private async createNodeExecuteContext({ - nodeId, - nodeName, - stepConfig, - stepContext: stepCtx, - itemsByConnection, - }: CreateNodeExecuteContextParams) { - const node: INode = { - id: nodeId, - name: nodeName, - type: stepConfig.nodeType, - typeVersion: stepConfig.typeVersion, - position: [0, 0], - parameters: stepConfig.parameters, - continueOnFail: stepConfig.continueOnFail, - }; - - const workflow = new Workflow({ - id: stepCtx.workflowId, - nodes: [node], - connections: {}, - active: false, - nodeTypes: this.deps.nodeTypes, - }); - - const firstInput = itemsByConnection[0] ?? []; - const connectionInputData = firstInput.length > 0 ? firstInput : [{ json: {} }]; - const inputData: ITaskDataConnections = { - main: [connectionInputData, ...itemsByConnection.slice(1)], - }; - - const runExecutionData = createRunExecutionData({ - startData: {}, - resultData: { runData: {} }, - executionData: { - contextData: {}, - nodeExecutionStack: [], - metadata: {}, - waitingExecution: {}, - waitingExecutionSource: null, - }, - }); - - const executeData: IExecuteData = { node, data: inputData, source: null }; - const mode: WorkflowExecuteMode = stepCtx.mode === 'production' ? 'trigger' : 'manual'; - const additionalData = await this.deps.additionalDataFactory(stepCtx.executionId); - - return new ExecuteContext( - workflow, - node, - additionalData, - mode, - runExecutionData, - 0, - connectionInputData, - inputData, - executeData, - [], - ); - } - /** * Runs the node and settles its cleanup functions. * @@ -150,21 +106,18 @@ export class V1StepExecutor implements IStepExecutor { * failure with `continueOnFail` passes the input items through as output. * A null result yields no output. */ - private async runNode({ - nodeType, - nodeExecuteContext, - }: RunNodeParams): Promise { + private async runNode({ nodeType, context }: RunNodeParams): Promise { let result: NodeRunResult; try { const value = isNodeClassInstance(nodeType) - ? await nodeType.execute(nodeExecuteContext) - : await nodeType.execute.call(nodeExecuteContext); + ? await nodeType.execute(context) + : await nodeType.execute.call(context); result = { ok: true, value }; } catch (error) { result = { ok: false, error }; } - const { closeFunctions } = nodeExecuteContext; + const { closeFunctions } = context; if (closeFunctions.length > 0) { const closeResults = await Promise.allSettled(closeFunctions.map(async (fn) => await fn())); if (result.ok) { @@ -175,21 +128,21 @@ export class V1StepExecutor implements IStepExecutor { throw rejected.reason instanceof Error ? rejected.reason : new UnexpectedError("Error on node's close function", { - extra: { nodeName: nodeExecuteContext.getNode().name }, + extra: { nodeName: context.getNode().name }, }); } } } if (!result.ok) { - if (!nodeExecuteContext.continueOnFail()) throw result.error; - return [nodeExecuteContext.getInputData()]; + if (!context.continueOnFail()) throw result.error; + return [context.getInputData()]; } if (result.value === null || result.value === undefined) return []; if (Array.isArray(result.value)) return result.value as INodeExecutionData[][]; - throw new EngineRequestNotSupportedError(nodeExecuteContext.getNode().type); + throw new EngineRequestNotSupportedError(context.getNode().type); } } diff --git a/packages/@n8n/node-engine-compatibility/src/v1-workflow-converter.ts b/packages/@n8n/node-engine-compatibility/src/v1-workflow-converter.ts index 9ab0c580fd4..8ba153067db 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-workflow-converter.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-workflow-converter.ts @@ -1,6 +1,7 @@ import type { GraphEdge, GraphNode, WorkflowGraph } from '@n8n/engine'; import type { INode, INodeConnections, IWorkflowBase } from 'n8n-workflow'; +import { MAIN_CONNECTION_TYPE, MANUAL_TRIGGER_TYPE, SPLIT_IN_BATCHES_TYPE } from './constants'; import { UnsupportedConnectionTypeError, UnsupportedCycleError, @@ -10,10 +11,6 @@ import { } from './errors'; import type { V1NodeStepConfig } from './types'; -const MANUAL_TRIGGER_TYPE = 'n8n-nodes-base.manualTrigger'; -const SPLIT_IN_BATCHES_TYPE = 'n8n-nodes-base.splitInBatches'; -const MAIN_CONNECTION_TYPE = 'main'; - /** * Common v1 trigger types that don't end in "Trigger". Non-exhaustive: combined * with the name heuristic below, it exists to reject unsupported triggers with a diff --git a/packages/@n8n/node-engine-compatibility/vitest.config.ts b/packages/@n8n/node-engine-compatibility/vitest.config.ts index c312d4bbd0a..bd8848dab9b 100644 --- a/packages/@n8n/node-engine-compatibility/vitest.config.ts +++ b/packages/@n8n/node-engine-compatibility/vitest.config.ts @@ -2,4 +2,5 @@ import { createVitestConfig } from '@n8n/vitest-config/node'; export default createVitestConfig({ exclude: ['**/node_modules/**', '**/dist/**'], + setupFiles: ['./src/__tests__/setup-vm-evaluator.ts'], }); diff --git a/packages/workflow/src/expression.ts b/packages/workflow/src/expression.ts index a7c54798a6a..9e83f6747cf 100644 --- a/packages/workflow/src/expression.ts +++ b/packages/workflow/src/expression.ts @@ -289,6 +289,21 @@ export class Expression { if (Expression.vmEvaluator) await Expression.vmEvaluator.release(this); } + async withIsolate(fn: () => Promise): Promise { + const acquired = await this.acquireIsolate(); + try { + return await fn(); + } finally { + if (acquired) { + try { + await this.releaseIsolate(); + } catch (error) { + LoggerProxy.error('Failed to release expression isolate', { error }); + } + } + } + } + /** * Dispose the VM evaluator and release resources. * Should be called during application shutdown or test teardown. diff --git a/packages/workflow/src/workflow-expression.ts b/packages/workflow/src/workflow-expression.ts index 0479a4e4a1b..c8b2d6c55c1 100644 --- a/packages/workflow/src/workflow-expression.ts +++ b/packages/workflow/src/workflow-expression.ts @@ -278,4 +278,8 @@ export class WorkflowExpression { async releaseIsolate(): Promise { await this.expression.releaseIsolate(); } + + async withIsolate(fn: () => Promise): Promise { + return await this.expression.withIsolate(fn); + } } diff --git a/packages/workflow/test/expression-with-isolate.test.ts b/packages/workflow/test/expression-with-isolate.test.ts new file mode 100644 index 00000000000..1b2e0d0959c --- /dev/null +++ b/packages/workflow/test/expression-with-isolate.test.ts @@ -0,0 +1,55 @@ +import { Expression } from '../src/expression'; + +describe('Expression.withIsolate', () => { + const expression = new Expression('Europe/Berlin'); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('returns the result of fn', async () => { + vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true); + const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue(); + + await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe( + 'result', + ); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('releases when fn throws, propagating the error', async () => { + vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true); + const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue(); + + await expect( + expression.withIsolate(async () => { + return await Promise.reject(new Error('boom from fn')); + }), + ).rejects.toThrow('boom from fn'); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('does not release an isolate it did not acquire', async () => { + vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(false); + const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue(); + + await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe( + 'result', + ); + expect(release).not.toHaveBeenCalled(); + }); + + it('never masks the outcome of fn with a release failure', async () => { + vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true); + vi.spyOn(expression, 'releaseIsolate').mockRejectedValue(new Error('release boom')); + + await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe( + 'result', + ); + await expect( + expression.withIsolate(async () => { + return await Promise.reject(new Error('boom from fn')); + }), + ).rejects.toThrow('boom from fn'); + }); +});