diff --git a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts index c0d060b8fdc..f9d8cdc23ce 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts @@ -203,6 +203,45 @@ describe('step execution (integration)', () => { expect(execution.finishedAt).toBeInstanceOf(Date); }); + it('runs a fan-in once, with each input slot fed by its branch', async () => { + const diamondGraph: WorkflowGraph = { + nodes: [ + { id: 'trigger', name: 'Webhook', type: 'trigger' }, + { id: 'node-a', name: 'A', type: 'v1-node' }, + { id: 'node-b', name: 'B', type: 'v1-node' }, + { id: 'node-m', name: 'M', type: 'v1-node' }, + ], + edges: [ + { from: 'trigger', to: 'node-a', outputIndex: 0, inputIndex: 0 }, + { from: 'trigger', to: 'node-b', outputIndex: 0, inputIndex: 0 }, + { from: 'node-a', to: 'node-m', outputIndex: 0, inputIndex: 0 }, + { from: 'node-b', to: 'node-m', outputIndex: 0, inputIndex: 1 }, + ], + }; + const requests: StepExecutionRequest[] = []; + const executor: IStepExecutor = { + execute: async (request) => { + requests.push(request); + await Promise.resolve(); + return { outputs: [[{ json: { ran: request.node.id } }]] }; + }, + }; + + const { execution } = await runWorkflow( + executor, + { body: { name: 'ada' } }, + { workflowId: 'wf-diamond', graph: diamondGraph }, + ); + + // the merge ran exactly once, after both branches, one slot per branch + const merge = requests.filter(({ node }) => node.id === 'node-m'); + expect(merge).toHaveLength(1); + expect(merge[0].inputs).toEqual([[{ json: { ran: 'node-a' } }], [{ json: { ran: 'node-b' } }]]); + + expect(execution.status).toBe('completed'); + expect(execution.finishedAt).toBeInstanceOf(Date); + }); + it('is idempotent across duplicate step:ready deliveries', async () => { const { executionStore, stepStore } = stores(); const execute = vi.fn().mockResolvedValue({ outputs: [[{ json: { n: 1 } }]] }); diff --git a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts index fdd200c2d08..621bd90f882 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts @@ -133,6 +133,111 @@ describe('StepReadyHandler', () => { ); }); + it('gathers a fan-in: each input slot from its predecessor', async () => { + // trigger fans out to a and b, which converge on m's two input slots + const diamond: WorkflowGraph = { + nodes: [...graph.nodes, { id: 'm', name: 'M', type: 'v1-node' }], + edges: [ + { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 0 }, + { from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 }, + { from: 'a', to: 'm', outputIndex: 0, inputIndex: 0 }, + { from: 'b', to: 'm', outputIndex: 0, inputIndex: 1 }, + ], + }; + const stepStore = makeStepStore( + { id: 'step-m', nodeId: 'm' }, + { + loadStepOutputs: vi.fn().mockResolvedValue({ + a: [[{ json: { from: 'a' } }]], + b: [[{ json: { from: 'b' } }]], + }), + }, + ); + const executor = makeExecutor(); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: diamond }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await handler.handle({ ...event, stepId: 'step-m' }); + + // one round trip for all predecessors + expect(stepStore.loadStepOutputs).toHaveBeenCalledTimes(1); + expect(stepStore.loadStepOutputs).toHaveBeenCalledWith('exec-1', ['a', 'b']); + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ inputs: [[{ json: { from: 'a' } }], [{ json: { from: 'b' } }]] }), + ); + }); + + it('leaves an input slot no edge feeds null', async () => { + const gapped: WorkflowGraph = { + nodes: [...graph.nodes, { id: 'm', name: 'M', type: 'v1-node' }], + edges: [ + { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 0 }, + { from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 }, + { from: 'a', to: 'm', outputIndex: 0, inputIndex: 0 }, + { from: 'b', to: 'm', outputIndex: 0, inputIndex: 2 }, + ], + }; + const stepStore = makeStepStore( + { id: 'step-m', nodeId: 'm' }, + { + loadStepOutputs: vi.fn().mockResolvedValue({ + a: [[{ json: { from: 'a' } }]], + b: [[{ json: { from: 'b' } }]], + }), + }, + ); + const executor = makeExecutor(); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: gapped }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await handler.handle({ ...event, stepId: 'step-m' }); + + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ + inputs: [[{ json: { from: 'a' } }], null, [{ json: { from: 'b' } }]], + }), + ); + }); + + it('feeds two input slots from one predecessor wired to both', async () => { + const doubled: WorkflowGraph = { + nodes: [...graph.nodes, { id: 'm', name: 'M', type: 'v1-node' }], + edges: [ + { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 0 }, + { from: 'a', to: 'm', outputIndex: 0, inputIndex: 0 }, + { from: 'a', to: 'm', outputIndex: 0, inputIndex: 1 }, + ], + }; + const stepStore = makeStepStore( + { id: 'step-m', nodeId: 'm' }, + { loadStepOutputs: vi.fn().mockResolvedValue({ a: [[{ json: { from: 'a' } }]] }) }, + ); + const executor = makeExecutor(); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: doubled }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await handler.handle({ ...event, stepId: 'step-m' }); + + expect(stepStore.loadStepOutputs).toHaveBeenCalledWith('exec-1', ['a']); + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ + inputs: [[{ json: { from: 'a' } }], [{ json: { from: 'a' } }]], + }), + ); + }); + it('fails the step when the executor produces more than one output slot', async () => { // A single-wired If passes graph validation but still emits two slots — // this guard is where branch selection gets rejected until it exists. @@ -200,26 +305,27 @@ describe('StepReadyHandler', () => { expect(stepStore.completeStep).toHaveBeenCalledWith('step-b', [null]); }); - it('throws when a predecessor row carries more than one output slot', async () => { - // the write-time guard makes this unreachable; reaching it means the store - // and this handler disagree, so nothing is recorded + it('reads the slot the edge names from a predecessor row carrying more slots', async () => { + // write-time guards keep rows single-slot today; the read side doesn't + // re-enforce that, it just takes the edge's slot const stepStore = makeStepStore( { id: 'step-b', nodeId: 'b' }, - { loadStepOutputs: vi.fn().mockResolvedValue({ a: [[{ json: {} }], [{ json: {} }]] }) }, + { + loadStepOutputs: vi + .fn() + .mockResolvedValue({ a: [[{ json: { slot: 0 } }], [{ json: { slot: 1 } }]] }), + }, ); const executor = makeExecutor(); const handler = new StepReadyHandler(makeExecutionStore(), stepStore, makeQueue(), { v1StepExecutor: executor, }); - await expect(handler.handle({ ...event, stepId: 'step-b' })).rejects.toMatchObject({ - name: 'UnexpectedError', - message: expect.stringContaining('more than one output slot') as string, - }); + await handler.handle({ ...event, stepId: 'step-b' }); - expect(executor.execute).not.toHaveBeenCalled(); - expect(stepStore.completeStep).not.toHaveBeenCalled(); - expect(stepStore.failStep).not.toHaveBeenCalled(); + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ inputs: [[{ json: { slot: 0 } }]] }), + ); }); it('throws, running nothing, when the predecessor step has no completed outputs', async () => { @@ -440,34 +546,19 @@ describe('StepReadyHandler', () => { */ it.each([ { - reason: 'more than one edge feeds it (fan-in)', + reason: 'two edges feed the same input slot', stepId: 'step-b', steps: () => makeStepStore({ id: 'step-b', nodeId: 'b' }), execution: () => makeExecutionStore({ graph: { nodes: graph.nodes, - // second input into b makes it a fan-in - edges: [...graph.edges, { from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 1 }], + // both a and the trigger feed b's slot 0 + edges: [...graph.edges, { from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 }], }, }), deps: (executor: IStepExecutor): ExternalDependencies => ({ v1StepExecutor: executor }), - expected: { name: 'UnexpectedError', message: 'incoming edges' }, - }, - { - reason: 'one node feeds it through two edges', - stepId: 'step-b', - steps: () => makeStepStore({ id: 'step-b', nodeId: 'b' }), - execution: () => - makeExecutionStore({ - graph: { - nodes: graph.nodes, - // 'a' connected to b twice still counts per edge, not per node - edges: [...graph.edges, { from: 'a', to: 'b', outputIndex: 1, inputIndex: 1 }], - }, - }), - deps: (executor: IStepExecutor): ExternalDependencies => ({ v1StepExecutor: executor }), - expected: { name: 'UnexpectedError', message: 'incoming edges' }, + expected: { name: 'UnexpectedError', message: 'input slot' }, }, { reason: "its edge leaves the predecessor's second output", @@ -479,13 +570,13 @@ describe('StepReadyHandler', () => { nodes: graph.nodes, edges: [ graph.edges[0], - // slot routing beyond 0→0 is rejected at graph validation + // output slots beyond 0 are rejected at graph validation { from: 'a', to: 'b', outputIndex: 1, inputIndex: 0 }, ], }, }), deps: (executor: IStepExecutor): ExternalDependencies => ({ v1StepExecutor: executor }), - expected: { name: 'UnexpectedError', message: 'slot 0' }, + expected: { name: 'UnexpectedError', message: 'output slot' }, }, { reason: 'its node has no predecessor in the graph', diff --git a/packages/@n8n/engine/src/execution/execution.types.ts b/packages/@n8n/engine/src/execution/execution.types.ts index 76de5b06ab8..9708a9a5e9d 100644 --- a/packages/@n8n/engine/src/execution/execution.types.ts +++ b/packages/@n8n/engine/src/execution/execution.types.ts @@ -15,7 +15,7 @@ export type StepStatus = 'queued' | 'running' | 'completed' | 'failed' | 'cancel * input slot; the engine understands the slot structure and nothing below it — * a slot's contents are opaque and step-type-specific. * - * TODO(CAT-2874): only slot 0 is populated until multi-slot routing lands; - * graph validation rejects edges that use any other slot. + * TODO(CAT-2874): outputs currently only support slot 0 (graph validation rejects edges with `outputIndex !== 0`). + * Inputs may populate multiple slots; an unfilled slot is represented as `null`. */ export type StepSlots = JsonValue[]; diff --git a/packages/@n8n/engine/src/execution/step-ready-handler.ts b/packages/@n8n/engine/src/execution/step-ready-handler.ts index aa921076e85..813961597bd 100644 --- a/packages/@n8n/engine/src/execution/step-ready-handler.ts +++ b/packages/@n8n/engine/src/execution/step-ready-handler.ts @@ -1,6 +1,6 @@ import { UnexpectedError, UnimplementedError } from '../common'; import type { ExternalDependencies, IStepExecutor } from '../dependencies'; -import type { GraphNode } from '../graph'; +import type { GraphEdge, GraphNode } from '../graph'; import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; import type { StepSlots } from './execution.types'; @@ -129,48 +129,34 @@ export class StepReadyHandler { } } - /** Inputs for `node`, taken from its predecessor's output. */ + /** Inputs for `node`, each slot taken from its predecessor's output. */ private async gatherInputs(execution: ExecutionRecord, step: StepRecord): Promise { - const incoming = execution.graph.edges.filter((edge) => edge.to === step.nodeId); - if (incoming.length === 0) { + // These are all the edges that feed into the node this step runs. + const incomingEdges = execution.graph.edges.filter((edge) => edge.to === step.nodeId); + if (incomingEdges.length === 0) { // Steps are planned only for a completed step's successors, so a step // without a predecessor means the graph and the step rows disagree. throw new UnexpectedError( `step ${step.id} runs node ${step.nodeId}, which has no predecessor in the execution graph`, ); } - // TODO(CAT-2874): support multiple inputs. We should have rejected this - // graph at validation time. - if (incoming.length > 1) { - throw new UnexpectedError( - `step ${step.id} runs node ${step.nodeId}, which has ${incoming.length} incoming edges; validated graphs have at most one edge per input slot`, - ); - } - const [edge] = incoming; - // TODO(CAT-2874): route by slot indices. We should have rejected this - // graph at validation time. - if (edge.outputIndex !== 0 || edge.inputIndex !== 0) { - throw new UnexpectedError( - `step ${step.id} runs node ${step.nodeId} through edge slots ${edge.outputIndex} → ${edge.inputIndex}; validated graphs only use slot 0`, - ); + + validateIncomingEdges(incomingEdges, step); + + const incomingOutputsByNodeId = await this.loadIncomingOutputs(execution.id, incomingEdges); + + // Array of length equal to the highest input slot plus one. + // The entries are `null` placeholders filled by the loop immediately below. + const inputs: StepSlots = Array.from( + { length: Math.max(...incomingEdges.map((edge) => edge.inputIndex)) + 1 }, + () => null, + ); + + for (const edge of incomingEdges) { + populateInputFromPredecessor(edge, step, incomingOutputsByNodeId, inputs); } - const outputsByNodeId = await this.stepStore.loadStepOutputs(execution.id, [edge.from]); - const predecessorOutputs = outputsByNodeId[edge.from]; - if (!predecessorOutputs) { - // A step is planned only once every predecessor completed, so running on - // a fabricated empty input would mask a planner/store inconsistency. - throw new UnexpectedError( - `step ${step.id} reads node ${edge.from}, whose step has not completed`, - ); - } - if (predecessorOutputs.length > 1) { - throw new UnexpectedError( - `step ${step.id} reads node ${edge.from}, whose step recorded more than one output slot; the write-time guard admits only slot 0`, - ); - } - - return [predecessorOutputs[0] ?? null]; + return inputs; } /** @@ -190,6 +176,60 @@ export class StepReadyHandler { throw new UnimplementedError(`step ${step.id}: no executor for step type ${node.type}`); } + + /** + * Loads the outputs of all the source steps of `incomingEdges`, keyed by the source node ID. + * The source steps must have completed. + * @param executionId to gather the predecessor outputs + * @param incomingEdges to know which predecessor nodes to gather outputs from + * @returns + */ + private async loadIncomingOutputs( + executionId: string, + incomingEdges: GraphEdge[], + ): Promise> { + const predecessorNodeIds = [...new Set(incomingEdges.map((edge) => edge.from))]; + return await this.stepStore.loadStepOutputs(executionId, predecessorNodeIds); + } +} + +// Validate that the incoming edge meets our constraints. filledSlots tracks the filled +// input slots so we can detect multiple edges into the same slot. +function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): void { + const filledSlots: Set = new Set(); + for (const edge of incomingEdges) { + // TODO(CAT-2874): route from non-zero output slots. We should have + // rejected this graph at validation time. + if (edge.outputIndex !== 0) { + throw new UnexpectedError( + `step ${step.id} runs node ${step.nodeId}, fed from output slot ${edge.outputIndex} of node ${edge.from}; validated graphs only use output slot 0`, + ); + } + if (filledSlots.has(edge.inputIndex)) { + // We should have rejected this graph at validation time. + throw new UnexpectedError( + `step ${step.id} runs node ${step.nodeId}, which has more than one edge into input slot ${edge.inputIndex}; validated graphs have at most one edge per input slot`, + ); + } + filledSlots.add(edge.inputIndex); + } +} + +function populateInputFromPredecessor( + edge: GraphEdge, + step: StepRecord, + incomingOutputsByNodeId: Record, + inputs: StepSlots, +): void { + const predecessorOutputs = incomingOutputsByNodeId[edge.from]; + if (!predecessorOutputs) { + // A step is planned only once every predecessor completed, so running on + // a fabricated empty input would mask a planner/store inconsistency. + throw new UnexpectedError( + `step ${step.id} reads node ${edge.from}, whose step has not completed`, + ); + } + inputs[edge.inputIndex] = predecessorOutputs[edge.outputIndex] ?? null; } function toStepError(error: unknown): StepError { diff --git a/packages/@n8n/engine/src/graph/__tests__/validate-executable-graph.test.ts b/packages/@n8n/engine/src/graph/__tests__/validate-executable-graph.test.ts index 154ce1dc606..6120948600e 100644 --- a/packages/@n8n/engine/src/graph/__tests__/validate-executable-graph.test.ts +++ b/packages/@n8n/engine/src/graph/__tests__/validate-executable-graph.test.ts @@ -46,6 +46,28 @@ describe('validateExecutableGraph', () => { expect(() => validateExecutableGraph(graph)).not.toThrow(); }); + it.each([ + { slot: 'input', edge: { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: -1 } }, + { slot: 'input', edge: { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 1.5 } }, + { slot: 'input', edge: { from: 'trigger', to: 'a', outputIndex: 0, inputIndex: NaN } }, + { slot: 'output', edge: { from: 'trigger', to: 'a', outputIndex: NaN, inputIndex: 0 } }, + ])('rejects an edge whose $slot slot index is $edge', ({ edge }) => { + const graph: WorkflowGraph = { nodes: validGraph.nodes, edges: [edge] }; + + expect(() => validateExecutableGraph(graph)).toThrow(GraphValidationError); + expect(() => validateExecutableGraph(graph)).toThrow('non-negative integer'); + }); + + it('rejects an edge whose input slot index is above 100', () => { + const graph: WorkflowGraph = { + nodes: validGraph.nodes, + edges: [{ from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 101 }], + }; + + expect(() => validateExecutableGraph(graph)).toThrow(GraphValidationError); + expect(() => validateExecutableGraph(graph)).toThrow('above 100'); + }); + it('rejects an edge leaving an output slot other than 0 as unimplemented', () => { const graph: WorkflowGraph = { nodes: [...validGraph.nodes, { id: 'b', name: 'B', type: 'v1-node' }], @@ -56,14 +78,22 @@ describe('validateExecutableGraph', () => { expect(() => validateExecutableGraph(graph)).toThrow('output slot'); }); - it('rejects an edge arriving at an input slot other than 0 as unimplemented', () => { + it('accepts a fan-in: edges into distinct input slots of one node', () => { const graph: WorkflowGraph = { - nodes: [...validGraph.nodes, { id: 'b', name: 'B', type: 'v1-node' }], - edges: [...validGraph.edges, { from: 'a', to: 'b', outputIndex: 0, inputIndex: 1 }], + nodes: [ + ...validGraph.nodes, + { id: 'b', name: 'B', type: 'v1-node' }, + { id: 'm', name: 'M', type: 'v1-node' }, + ], + edges: [ + ...validGraph.edges, + { from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 }, + { from: 'a', to: 'm', outputIndex: 0, inputIndex: 0 }, + { from: 'b', to: 'm', outputIndex: 0, inputIndex: 1 }, + ], }; - expect(() => validateExecutableGraph(graph)).toThrow(UnimplementedError); - expect(() => validateExecutableGraph(graph)).toThrow('input slot'); + expect(() => validateExecutableGraph(graph)).not.toThrow(); }); it('rejects two edges into the same input slot of one node as unimplemented', () => { diff --git a/packages/@n8n/engine/src/graph/validate-executable-graph.ts b/packages/@n8n/engine/src/graph/validate-executable-graph.ts index 349a1ccdb0d..7ce4fae5dba 100644 --- a/packages/@n8n/engine/src/graph/validate-executable-graph.ts +++ b/packages/@n8n/engine/src/graph/validate-executable-graph.ts @@ -1,6 +1,8 @@ import { UnimplementedError } from '../common'; import type { WorkflowGraph } from './workflow-graph'; +const MAX_SLOT_INDEX = 100; + /** Thrown when a graph fails a structural rule and can never execute. */ export class GraphValidationError extends Error { constructor(message: string) { @@ -32,19 +34,31 @@ export function validateExecutableGraph(graph: WorkflowGraph): void { throw new UnimplementedError('Graphs with back-edges (loops) are not supported yet'); } - // TODO(CAT-2874): multi-slot routing arrives with branching; until then only - // slot 0 → slot 0 edges run, and the runtime can assume single-slot IO. + // Slot indices are structural, so they're enforced here rather than left to + // the transport boundary. TODO(CAT-3042): enforce an upper bound too. + for (const edge of graph.edges) { + for (const index of [edge.outputIndex, edge.inputIndex]) { + if (!Number.isInteger(index) || index < 0) { + throw new GraphValidationError( + `Edge ${edge.from} → ${edge.to} has slot index ${index}; slot indices are non-negative integers`, + ); + } + if (index > MAX_SLOT_INDEX) { + throw new GraphValidationError( + `Edge ${edge.from} → ${edge.to} has slot index ${index}; slot indices above ${MAX_SLOT_INDEX} are not supported yet`, + ); + } + } + } + + // TODO(CAT-2874): multi-slot outputs arrive with branching; until then only + // output slot 0 fires, and the runtime can assume single-slot outputs. for (const edge of graph.edges) { if (edge.outputIndex !== 0) { throw new UnimplementedError( `Edge ${edge.from} → ${edge.to} leaves output slot ${edge.outputIndex}; only output slot 0 is supported yet`, ); } - if (edge.inputIndex !== 0) { - throw new UnimplementedError( - `Edge ${edge.from} → ${edge.to} arrives at input slot ${edge.inputIndex}; only input slot 0 is supported yet`, - ); - } } // TODO(CAT-3982): same-slot convergence gets a defined meaning (concatenation);