From f3db19df76fa53c8ac44c3e1e248f1bcd3030aed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Wed, 9 Sep 2026 11:30:38 +0000 Subject: [PATCH] fix(core): Memoize pairedItem ancestry traversal (#38085) Co-authored-by: Claude Fable 5 Co-authored-by: Emilia <100027345+sovietspaceship@users.noreply.github.com> --- .../benchmarks/paired-item/traversal.bench.ts | 250 ++++++++++++ .../workflow-data-proxy-paired-item-memo.ts | 42 ++ packages/workflow/src/workflow-data-proxy.ts | 45 ++- ...rkflow-data-proxy-paired-item-memo.test.ts | 55 +++ .../workflow/test/workflow-data-proxy.test.ts | 359 ++++++++++++++++++ 5 files changed, 749 insertions(+), 2 deletions(-) create mode 100644 packages/testing/performance/benchmarks/paired-item/traversal.bench.ts create mode 100644 packages/workflow/src/workflow-data-proxy-paired-item-memo.ts create mode 100644 packages/workflow/test/workflow-data-proxy-paired-item-memo.test.ts diff --git a/packages/testing/performance/benchmarks/paired-item/traversal.bench.ts b/packages/testing/performance/benchmarks/paired-item/traversal.bench.ts new file mode 100644 index 00000000000..f153534a22b --- /dev/null +++ b/packages/testing/performance/benchmarks/paired-item/traversal.bench.ts @@ -0,0 +1,250 @@ +/** + * Paired item ancestry traversal benchmarks. + * + * `$('Node').item` / `.itemMatching()` walk an item's `pairedItem` chain back + * through every ancestor, once per expression that uses them. Ancestry is a DAG, + * so an item reachable through many paths must be expanded once, not once per path. + * + * Keep the chains short: `itemMatching` also runs a `getParentNodes` connection + * check that is superlinear in the chain length and would otherwise dominate. + */ +import { bench, describe } from 'vitest'; +import { Workflow, WorkflowDataProxy } from 'n8n-workflow'; +import type { + IConnections, + IExecuteData, + INode, + INodeExecutionData, + INodeTypes, + IPairedItemData, + IRunExecutionData, + ITaskData, +} from 'n8n-workflow'; + +import { BENCH_OPTIONS } from '../bench-options'; + +const MAIN = 'main'; + +/** An unknown node type leaves the node as given; the traversal reads only run data and connections. */ +const nodeTypes: INodeTypes = { + getByName: () => undefined as never, + getByNameAndVersion: () => undefined as never, + getKnownTypes: () => ({}), +}; + +type ChainSpec = { + /** Items in each node's output; its length is the chain length. */ + widths: number[]; + pairing: (d: number, i: number, widths: number[]) => IPairedItemData | IPairedItemData[]; +}; + +/** A -> B -> C -> ... with run data attached to every node. */ +function buildChain({ widths, pairing }: ChainSpec) { + const nodes: INode[] = []; + const connections: IConnections = {}; + const runData: IRunExecutionData['resultData']['runData'] = {}; + + for (let d = 0; d < widths.length; d++) { + const name = `n${d}`; + nodes.push({ + id: name, + name, + type: 'n8n-nodes-base.noOp', + typeVersion: 1, + position: [d * 100, 0], + parameters: {}, + }); + + if (d < widths.length - 1) { + connections[name] = { [MAIN]: [[{ node: `n${d + 1}`, type: MAIN, index: 0 }]] }; + } + + const items: INodeExecutionData[] = []; + for (let i = 0; i < widths[d]!; i++) { + const item: INodeExecutionData = { json: { d, i } }; + if (d > 0) item.pairedItem = pairing(d, i, widths); + items.push(item); + } + + runData[name] = [ + { + startTime: 0, + executionTime: 0, + executionIndex: d, + source: d === 0 ? [] : [{ previousNode: `n${d - 1}`, previousNodeRun: 0 }], + data: { [MAIN]: [items] }, + } satisfies ITaskData, + ]; + } + + return { nodes, connections, runData }; +} + +/** A proxy on the last node of the chain, ready to trace back to the first. */ +function proxyForChain(spec: ChainSpec) { + const { nodes, connections, runData } = buildChain(spec); + const activeNode = `n${spec.widths.length - 1}`; + const taskData = runData[activeNode]![0]!; + const inputData = taskData.data![MAIN]![0]!; + + const executeData: IExecuteData = { + data: taskData.data!, + node: nodes[nodes.length - 1]!, + source: { [MAIN]: taskData.source }, + }; + + return new WorkflowDataProxy( + new Workflow({ id: 'bench', name: 'bench', nodes, connections, active: false, nodeTypes }), + { resultData: { runData } } as IRunExecutionData, + 0, + 0, + activeNode, + inputData, + {}, + 'manual', + {}, + executeData, + ).getDataProxy(); +} + +/** The last node's items carry a single paired item, the proxy's entry point into the walk. */ +const isLastNode = (d: number, widths: number[]) => d === widths.length - 1; + +/** + * origin -> {bN, cN} -> mN -> {bN+1, cN+1} -> ... : 2^diamonds paths to the last + * merge, where the chain shapes above reach a shared ancestor from one node. + */ +function diamondProxy(diamonds: number) { + const nodes: INode[] = []; + const connections: IConnections = {}; + const runData: IRunExecutionData['resultData']['runData'] = {}; + + const addNode = (name: string) => { + nodes.push({ + id: name, + name, + type: 'n8n-nodes-base.noOp', + typeVersion: 1, + position: [nodes.length * 100, 0], + parameters: {}, + }); + }; + const link = (from: string, to: string, index: number) => { + connections[from] ??= { [MAIN]: [[]] }; + connections[from][MAIN][0]!.push({ node: to, type: MAIN, index }); + }; + const task = ( + source: ITaskData['source'], + pairedItem: IPairedItemData | IPairedItemData[] | undefined, + ) => + [ + { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source, + data: { [MAIN]: [[{ json: {}, ...(pairedItem ? { pairedItem } : {}) }]] }, + } satisfies ITaskData, + ] as ITaskData[]; + + addNode('origin'); + runData.origin = task([], undefined); + + let cur = 'origin'; + for (let i = 0; i < diamonds; i++) { + const [b, c, m] = [`b${i}`, `c${i}`, `m${i}`]; + for (const branch of [b, c]) { + addNode(branch); + link(cur, branch, 0); + runData[branch] = task([{ previousNode: cur, previousNodeRun: 0 }], { item: 0, input: 0 }); + } + addNode(m); + link(b, m, 0); + link(c, m, 1); + runData[m] = task( + [ + { previousNode: b, previousNodeRun: 0 }, + { previousNode: c, previousNodeRun: 0 }, + ], + [ + { item: 0, input: 0 }, + { item: 0, input: 1 }, + ], + ); + cur = m; + } + + addNode('end'); + link(cur, 'end', 0); + runData.end = task([{ previousNode: cur, previousNodeRun: 0 }], { item: 0, input: 0 }); + + const taskData = runData.end![0]!; + return new WorkflowDataProxy( + new Workflow({ id: 'bench', name: 'bench', nodes, connections, active: false, nodeTypes }), + { resultData: { runData } } as IRunExecutionData, + 0, + 0, + 'end', + taskData.data![MAIN]![0]!, + {}, + 'manual', + {}, + { + data: taskData.data!, + node: nodes[nodes.length - 1]!, + source: { [MAIN]: taskData.source }, + }, + ).getDataProxy(); +} + +describe('Paired item ancestry traversal', () => { + // Every item pairs to the same ancestor twice: 2^depth paths over `depth` + // distinct items. Exponential paths, must stay linear. + const recombining = proxyForChain({ + widths: Array(20).fill(1), + pairing: (d, i, widths) => + isLastNode(d, widths) + ? { item: i, input: 0 } + : [ + { item: 0, input: 0 }, + { item: 0, input: 0 }, + ], + }); + bench( + 'itemMatching: recombining chain (20 nodes, 2^20 paths)', + () => { + recombining.$('n0').itemMatching(0); + }, + BENCH_OPTIONS, + ); + + // One item paired to 10k inputs that all trace to the same origin: no path is + // walked twice, so every visit is new work. + const FAN = 10_000; + const fanIn = proxyForChain({ + widths: [1, FAN, 1, 1], + pairing: (d, i, widths) => { + if (isLastNode(d, widths)) return { item: i, input: 0 }; + if (d === 2) return Array.from({ length: FAN }, (_, k) => ({ item: k, input: 0 })); + return { item: 0, input: 0 }; + }, + }); + bench( + `itemMatching: fan-in (one item paired to ${FAN} ancestors)`, + () => { + fanIn.$('n0').itemMatching(0); + }, + BENCH_OPTIONS, + ); + + // Branches converge from two separate parents, so a shared ancestor is reached + // under one state from several routes. Exponential paths, must stay linear. + const diamonds = diamondProxy(14); + bench( + 'itemMatching: diamond chain (14 diamonds, 2^14 paths)', + () => { + diamonds.$('origin').itemMatching(0); + }, + BENCH_OPTIONS, + ); +}); diff --git a/packages/workflow/src/workflow-data-proxy-paired-item-memo.ts b/packages/workflow/src/workflow-data-proxy-paired-item-memo.ts new file mode 100644 index 00000000000..2ce7a6adcdb --- /dev/null +++ b/packages/workflow/src/workflow-data-proxy-paired-item-memo.ts @@ -0,0 +1,42 @@ +import type { INodeExecutionData, IPairedItemData, ISourceData } from './interfaces'; + +type PairedItemMemoResult = + | { ok: true; result: INodeExecutionData } + | { ok: false; error: unknown }; + +/** + * Memoizes one paired item traversal per (node run output, item) state. + * For a fixed destination, revisiting a state yields the same result or error. + */ +export class PairedItemMemo { + private readonly entries = new Map(); + + /** + * Resolves the memoized result for the given source data and paired item, + * or computes and memoizes a new result if none exists. + */ + resolve( + sourceData: ISourceData, + pairedItem: IPairedItemData, + compute: () => INodeExecutionData, + ): INodeExecutionData { + const key = this.toKey(sourceData, pairedItem); + + let entry = this.entries.get(key); + if (!entry) { + try { + entry = { ok: true, result: compute() }; + } catch (error) { + entry = { ok: false, error }; + } + this.entries.set(key, entry); + } + + if (entry.ok) return entry.result; + throw entry.error; + } + + private toKey(sourceData: ISourceData, pairedItem: IPairedItemData): string { + return `${sourceData.previousNodeRun ?? 0}|${sourceData.previousNodeOutput ?? 0}|${pairedItem.item}|${sourceData.previousNode}`; + } +} diff --git a/packages/workflow/src/workflow-data-proxy.ts b/packages/workflow/src/workflow-data-proxy.ts index 86b436d767d..0180b20e792 100644 --- a/packages/workflow/src/workflow-data-proxy.ts +++ b/packages/workflow/src/workflow-data-proxy.ts @@ -43,6 +43,7 @@ import type { Workflow } from './workflow'; import type { EnvProviderState } from './workflow-data-proxy-env-provider'; import { createEnvProvider, createEnvProviderState } from './workflow-data-proxy-env-provider'; import { getPinDataIfManualExecution } from './workflow-data-proxy-helpers'; +import { PairedItemMemo } from './workflow-data-proxy-paired-item-memo'; const isScriptingNode = (nodeName: string, workflow: Workflow) => { const node = workflow.getNode(nodeName); @@ -1034,7 +1035,26 @@ export class WorkflowDataProxy { incomingSourceData: ISourceData | null, initialPairedItem: IPairedItemData, usedMethodName: PairedItemMethod = PAIRED_ITEM_METHOD.$GET_PAIRED_ITEM, - nodeBeforeLast?: string, + ): INodeExecutionData => + resolvePairedItem( + destinationNodeName, + incomingSourceData, + initialPairedItem, + usedMethodName, + undefined, + // Ancestry is a DAG: branches recombine on shared ancestors (e.g. an + // Aggregate output pairing to all its inputs), so without memoization + // the walk revisits the same item exponentially often. + new PairedItemMemo(), + ); + + const resolvePairedItem = ( + destinationNodeName: string, + incomingSourceData: ISourceData | null, + initialPairedItem: IPairedItemData, + usedMethodName: PairedItemMethod, + nodeBeforeLast: string | undefined, + memo: PairedItemMemo, ): INodeExecutionData => { // Normalize inputs const [pairedItem, sourceData] = normalizeInputs(initialPairedItem, incomingSourceData); @@ -1043,6 +1063,26 @@ export class WorkflowDataProxy { throw createPairedItemNotFound(destinationNodeName, nodeBeforeLast); } + return memo.resolve(sourceData, pairedItem, () => + resolvePairedItemUncached( + destinationNodeName, + sourceData, + pairedItem, + usedMethodName, + nodeBeforeLast, + memo, + ), + ); + }; + + const resolvePairedItemUncached = ( + destinationNodeName: string, + sourceData: ISourceData, + pairedItem: IPairedItemData, + usedMethodName: PairedItemMethod, + nodeBeforeLast: string | undefined, + memo: PairedItemMemo, + ): INodeExecutionData => { const taskData = getTaskData(sourceData); const outputData = getNodeOutput(taskData, sourceData, nodeBeforeLast); const item = outputData[pairedItem.item]; @@ -1074,12 +1114,13 @@ export class WorkflowDataProxy { try { return createResultOk( - getPairedItem( + resolvePairedItem( destinationNodeName, nextSource, { ...nextPairedItem, input: inputIndex }, usedMethodName, sourceData.previousNode, + memo, ), ); } catch (error) { diff --git a/packages/workflow/test/workflow-data-proxy-paired-item-memo.test.ts b/packages/workflow/test/workflow-data-proxy-paired-item-memo.test.ts new file mode 100644 index 00000000000..ba2c1976f8c --- /dev/null +++ b/packages/workflow/test/workflow-data-proxy-paired-item-memo.test.ts @@ -0,0 +1,55 @@ +import type { INodeExecutionData } from '../src/interfaces'; +import { PairedItemMemo } from '../src/workflow-data-proxy-paired-item-memo'; + +describe('PairedItemMemo', () => { + const source = { previousNode: 'A', previousNodeRun: 1, previousNodeOutput: 2 }; + const item: INodeExecutionData = { json: {} }; + + test('computes once per state and replays the result', () => { + const memo = new PairedItemMemo(); + const compute = vi.fn(() => item); + + expect(memo.resolve(source, { item: 0 }, compute)).toBe(item); + expect(memo.resolve(source, { item: 0 }, compute)).toBe(item); + expect(compute).toHaveBeenCalledTimes(1); + }); + + test('replays the same error without recomputing', () => { + const memo = new PairedItemMemo(); + const error = new Error('boom'); + const compute = vi.fn(() => { + throw error; + }); + + expect(() => memo.resolve(source, { item: 0 }, compute)).toThrow(error); + expect(() => memo.resolve(source, { item: 0 }, compute)).toThrow(error); + expect(compute).toHaveBeenCalledTimes(1); + }); + + test('treats omitted run and output as 0', () => { + const memo = new PairedItemMemo(); + const compute = vi.fn(() => item); + + memo.resolve({ previousNode: 'A' }, { item: 0 }, compute); + memo.resolve( + { previousNode: 'A', previousNodeRun: 0, previousNodeOutput: 0 }, + { item: 0 }, + compute, + ); + expect(compute).toHaveBeenCalledTimes(1); + }); + + test.each([ + ['node', { ...source, previousNode: 'B' }, { item: 0 }], + ['run', { ...source, previousNodeRun: 0 }, { item: 0 }], + ['output', { ...source, previousNodeOutput: 0 }, { item: 0 }], + ['item', source, { item: 1 }], + ])('keys a different %s as a distinct state', (_, otherSource, otherItem) => { + const memo = new PairedItemMemo(); + const compute = vi.fn(() => item); + + memo.resolve(source, { item: 0 }, compute); + memo.resolve(otherSource, otherItem, compute); + expect(compute).toHaveBeenCalledTimes(2); + }); +}); diff --git a/packages/workflow/test/workflow-data-proxy.test.ts b/packages/workflow/test/workflow-data-proxy.test.ts index f66837e811a..2f9421ea2b2 100644 --- a/packages/workflow/test/workflow-data-proxy.test.ts +++ b/packages/workflow/test/workflow-data-proxy.test.ts @@ -9,6 +9,8 @@ import { type IDataObject, type IExecuteData, type INode, + type INodeExecutionData, + type IPairedItemData, type IPinData, type IRun, type IWorkflowBase, @@ -2264,3 +2266,360 @@ describe('WorkflowDataProxy', () => { }); }); }); + +describe('WorkflowDataProxy → pairedItem traversal through recombining ancestry', () => { + // Models a polling loop whose Aggregate-style node pairs its single output + // item to all of its inputs: ancestry branches fan out at every pass and + // recombine on the shared ancestor, forming a DAG. Traversal must resolve + // this in polynomial time - a tree walk would take ROWS^PASSES steps. + const ROWS = 8; + const PASSES = 25; + + const makeWorkflow = (): IWorkflowBase => ({ + id: '1', + name: 'test', + nodes: (['Start', 'Rows', 'Agg', 'End'] as const).map((name, i) => ({ + id: `uuid-${i}`, + name, + type: 'n8n-nodes-base.code', + typeVersion: 1, + position: [i * 100, 0] as [number, number], + parameters: {}, + })), + connections: { + Start: { main: [[{ node: 'Rows', type: NodeConnectionTypes.Main, index: 0 }]] }, + Rows: { main: [[{ node: 'Agg', type: NodeConnectionTypes.Main, index: 0 }]] }, + Agg: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] }, + }, + active: false, + activeVersionId: null, + isArchived: false, + createdAt: new Date(), + updatedAt: new Date(), + }); + + const taskData = (source: Array<{ previousNode: string; previousNodeRun?: number } | null>) => ({ + startTime: 0, + executionTime: 0, + executionIndex: 0, + source, + data: { main: [[]] as INodeExecutionData[][] }, + }); + + const makeRun = ( + startItems: INodeExecutionData[], + rowsPairing: (row: number) => number, + ): IRun => { + const start = taskData([null]); + start.data.main[0] = startItems; + + const rowsRuns = []; + const aggRuns = []; + for (let pass = 0; pass < PASSES; pass++) { + const rows = taskData([ + pass === 0 ? { previousNode: 'Start' } : { previousNode: 'Agg', previousNodeRun: pass - 1 }, + ]); + // only pass 0 is sourced from Start (multiple items); later passes are + // sourced from Agg, which emits a single item + rows.data.main[0] = Array.from({ length: ROWS }, (_, row) => ({ + json: { row }, + pairedItem: { item: pass === 0 ? rowsPairing(row) : 0 }, + })); + rowsRuns.push(rows); + + const agg = taskData([{ previousNode: 'Rows', previousNodeRun: pass }]); + agg.data.main[0] = [ + { json: { pass }, pairedItem: Array.from({ length: ROWS }, (_, item) => ({ item })) }, + ]; + aggRuns.push(agg); + } + + const end = taskData([{ previousNode: 'Agg', previousNodeRun: PASSES - 1 }]); + end.data.main[0] = [{ json: {}, pairedItem: { item: 0 } }]; + + return { + data: createRunExecutionData({ + resultData: { runData: { Start: [start], Rows: rowsRuns, Agg: aggRuns, End: [end] } }, + }), + mode: 'manual', + startedAt: new Date(), + status: 'success', + storedAt: 'db', + }; + }; + + test('resolves when all branches recombine on one ancestor item', () => { + const run = makeRun([{ json: { origin: true }, pairedItem: { item: 0 } }], () => 0); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(proxy.$('Start').item.json).toEqual({ origin: true }); + }); + + test('still detects ambiguity when branches resolve to different ancestor items', () => { + const startItems = Array.from({ length: ROWS }, (_, item) => ({ + json: { item }, + pairedItem: { item }, + })); + // each row pairs back to its own Start item, so resolution is ambiguous + const run = makeRun(startItems, (row) => row); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(() => proxy.$('Start').item).toThrowError('Multiple matches'); + }); +}); + +describe('WorkflowDataProxy → pairedItem traversal with sourceOverwrite routes', () => { + // One item pairs to the same ancestor node run twice: once through its task's + // default source and once through a `sourceOverwrite` redirect. The two routes + // must resolve as distinct states. + const makeWorkflow = (): IWorkflowBase => ({ + id: '1', + name: 'test', + nodes: (['Start', 'Mid', 'Join', 'End'] as const).map((name, i) => ({ + id: `uuid-${i}`, + name, + type: 'n8n-nodes-base.code', + typeVersion: 1, + position: [i * 100, 0] as [number, number], + parameters: {}, + })), + connections: { + Start: { main: [[{ node: 'Mid', type: NodeConnectionTypes.Main, index: 0 }]] }, + Mid: { main: [[{ node: 'Join', type: NodeConnectionTypes.Main, index: 0 }]] }, + Join: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] }, + }, + active: false, + activeVersionId: null, + isArchived: false, + createdAt: new Date(), + updatedAt: new Date(), + }); + + const midRun = (outputs: INodeExecutionData[][]) => ({ + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [{ previousNode: 'Start' }], + data: { main: outputs }, + }); + + // Mid run N emits one item paired to Start item `midRunPairing[N]` + const midRunsPairedTo = (midRunPairing: number[]) => + midRunPairing.map((startItem) => midRun([[{ json: {}, pairedItem: { item: startItem } }]])); + + const defaultJoinPairing: IPairedItemData[] = [ + { item: 0 }, + { item: 0, sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 1 } }, + ]; + + const makeRun = ( + startItems: INodeExecutionData[], + midRuns: ReturnType[], + joinPairing: IPairedItemData[] = defaultJoinPairing, + ): IRun => { + const start = { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [null], + data: { main: [startItems] }, + }; + + const join = { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [{ previousNode: 'Mid', previousNodeRun: 0 }], + data: { main: [[{ json: {}, pairedItem: joinPairing }]] }, + }; + + const end = { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [{ previousNode: 'Join' }], + data: { main: [[{ json: {}, pairedItem: { item: 0 } }]] }, + }; + + return { + data: createRunExecutionData({ + resultData: { runData: { Start: [start], Mid: midRuns, Join: [join], End: [end] } }, + }), + mode: 'manual', + startedAt: new Date(), + status: 'success', + storedAt: 'db', + }; + }; + + test('resolves when the default and overwritten routes agree on the ancestor item', () => { + const run = makeRun( + [{ json: { origin: true }, pairedItem: { item: 0 } }], + midRunsPairedTo([0, 0]), + ); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(proxy.$('Start').item.json).toEqual({ origin: true }); + }); + + test('detects ambiguity when the overwritten route reaches a different ancestor item', () => { + const startItems = [ + { json: { item: 0 }, pairedItem: { item: 0 } }, + { json: { item: 1 }, pairedItem: { item: 1 } }, + ]; + const run = makeRun(startItems, midRunsPairedTo([0, 1])); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(() => proxy.$('Start').item).toThrowError('Multiple matches'); + }); + + test('returns the resolving route when another route fails, also on its memoized revisit', () => { + const startItems = [{ json: { origin: true }, pairedItem: { item: 0 } }]; + // Mid run 1 has no pairedItem, so the overwritten route cannot resolve. + // It is listed twice so the second visit replays the memoized error. + const brokenRoute = { item: 0, sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 1 } }; + const midRuns = [...midRunsPairedTo([0]), midRun([[{ json: {} }]])]; + + const okFirst = getProxyFromFixture( + makeWorkflow(), + makeRun(startItems, midRuns, [{ item: 0 }, brokenRoute, brokenRoute]), + 'End', + ); + expect(okFirst.$('Start').item.json).toEqual({ origin: true }); + + const okLast = getProxyFromFixture( + makeWorkflow(), + makeRun(startItems, midRuns, [brokenRoute, brokenRoute, { item: 0 }]), + 'End', + ); + expect(okLast.$('Start').item.json).toEqual({ origin: true }); + + const allBroken = getProxyFromFixture( + makeWorkflow(), + makeRun(startItems, midRuns, [brokenRoute, brokenRoute]), + 'End', + ); + expect(() => allBroken.$('Start').item).toThrowError( + "Paired item data for item from node 'Mid' is unavailable", + ); + }); + + test('treats the same item index on different outputs of one run as distinct states', () => { + const startItems = [ + { json: { item: 0 }, pairedItem: { item: 0 } }, + { json: { item: 1 }, pairedItem: { item: 1 } }, + ]; + // Mid run 0 output 0 item 0 pairs to Start 0, output 1 item 0 pairs to Start 1 + const midRuns = [ + midRun([[{ json: {}, pairedItem: { item: 0 } }], [{ json: {}, pairedItem: { item: 1 } }]]), + ]; + const run = makeRun(startItems, midRuns, [ + { + item: 0, + sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 0, previousNodeOutput: 0 }, + }, + { + item: 0, + sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 0, previousNodeOutput: 1 }, + }, + ]); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(() => proxy.$('Start').item).toThrowError('Multiple matches'); + }); +}); + +describe('WorkflowDataProxy → pairedItem traversal through a diamond', () => { + // Start fans out to Left and Right, which join again in Merge. A Start item is + // reached under one state from two different parent nodes. + const makeWorkflow = (): IWorkflowBase => ({ + id: '1', + name: 'test', + nodes: (['Start', 'Left', 'Right', 'Merge', 'End'] as const).map((name, i) => ({ + id: `uuid-${i}`, + name, + type: 'n8n-nodes-base.code', + typeVersion: 1, + position: [i * 100, 0] as [number, number], + parameters: {}, + })), + connections: { + Start: { + main: [ + [ + { node: 'Left', type: NodeConnectionTypes.Main, index: 0 }, + { node: 'Right', type: NodeConnectionTypes.Main, index: 0 }, + ], + ], + }, + Left: { main: [[{ node: 'Merge', type: NodeConnectionTypes.Main, index: 0 }]] }, + Right: { main: [[{ node: 'Merge', type: NodeConnectionTypes.Main, index: 1 }]] }, + Merge: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] }, + }, + active: false, + activeVersionId: null, + isArchived: false, + createdAt: new Date(), + updatedAt: new Date(), + }); + + const task = (source: Array<{ previousNode: string } | null>, items: INodeExecutionData[]) => ({ + startTime: 0, + executionTime: 0, + executionIndex: 0, + source, + data: { main: [items] }, + }); + + const makeRun = ( + startItems: INodeExecutionData[], + leftItem: number, + rightItem: number, + ): IRun => ({ + data: createRunExecutionData({ + resultData: { + runData: { + Start: [task([null], startItems)], + Left: [task([{ previousNode: 'Start' }], [{ json: {}, pairedItem: { item: leftItem } }])], + Right: [ + task([{ previousNode: 'Start' }], [{ json: {}, pairedItem: { item: rightItem } }]), + ], + Merge: [ + task( + [{ previousNode: 'Left' }, { previousNode: 'Right' }], + [ + { + json: {}, + pairedItem: [ + { item: 0, input: 0 }, + { item: 0, input: 1 }, + ], + }, + ], + ), + ], + End: [task([{ previousNode: 'Merge' }], [{ json: {}, pairedItem: { item: 0 } }])], + }, + }, + }), + mode: 'manual', + startedAt: new Date(), + status: 'success', + storedAt: 'db', + }); + + test('resolves when both routes reach one ancestor item', () => { + const run = makeRun([{ json: { origin: true }, pairedItem: { item: 0 } }], 0, 0); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(proxy.$('Start').item.json).toEqual({ origin: true }); + }); + + test('detects ambiguity when the routes reach different ancestor items', () => { + const startItems = [0, 1].map((item) => ({ json: { item }, pairedItem: { item } })); + const run = makeRun(startItems, 0, 1); + const proxy = getProxyFromFixture(makeWorkflow(), run, 'End'); + + expect(() => proxy.$('Start').item).toThrowError('Multiple matches'); + }); +});