From 9d7abc2091bc7871b3c4451e54bfc34eb315c4f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Ovejero?= Date: Thu, 20 Aug 2026 11:31:28 +0000 Subject: [PATCH] feat(engine): Settle steps across loop iterations (no-changelog) (#36604) --- ...orkflow-step-execution.integration.test.ts | 22 +- .../engine/src/database/typeorm-step-store.ts | 67 ++++-- .../execution/__tests__/completion.test.ts | 78 +++++++ .../__tests__/execution-start-handler.test.ts | 2 +- .../execution/__tests__/loop-ledger.test.ts | 121 ++++++++++ .../execution/__tests__/settlement.test.ts | 205 +++++++++++++++-- .../__tests__/step-ready-handler.test.ts | 157 ++++++++++++- .../__tests__/step-settled-handler.test.ts | 8 +- .../@n8n/engine/src/execution/completion.ts | 42 ++++ .../@n8n/engine/src/execution/loop-ledger.ts | 72 ++++++ .../@n8n/engine/src/execution/settlement.ts | 216 ++++++++++++++---- .../src/execution/step-ready-handler.ts | 86 +++++-- .../src/execution/step-settled-handler.ts | 97 ++++---- .../@n8n/engine/src/execution/step-store.ts | 22 +- 14 files changed, 1028 insertions(+), 167 deletions(-) create mode 100644 packages/@n8n/engine/src/execution/__tests__/completion.test.ts create mode 100644 packages/@n8n/engine/src/execution/__tests__/loop-ledger.test.ts create mode 100644 packages/@n8n/engine/src/execution/completion.ts create mode 100644 packages/@n8n/engine/src/execution/loop-ledger.ts diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts index 0cf22272e6e..872a5d3217b 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts @@ -674,7 +674,7 @@ describe('workflow_step_execution table (integration)', () => { }); }); - it('TypeOrmStepStore.loadLatestStep returns the highest-iteration row, or null', async () => { + it("TypeOrmStepStore.loadLatestStepSummaries returns each node's highest-iteration row", async () => { const executionId = await createExecution(); const store = new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution)); await store.createSteps(executionId, [ @@ -685,10 +685,24 @@ describe('workflow_step_execution table (integration)', () => { const otherExecutionId = await createExecution(); await store.createSteps(otherExecutionId, [{ nodeId: 'b', iteration: 5, status: 'queued' }]); - const latest = await store.loadLatestStep(executionId, 'b'); - expect(latest).toMatchObject({ nodeId: 'b', iteration: 1, status: 'completed' }); + await store.createSteps(executionId, [ + { nodeId: 'c', iteration: 0, status: 'completed', outputs: [[{ json: {} }]] }, + ]); - expect(await store.loadLatestStep(executionId, 'ghost')).toBeNull(); + const latest = await store.loadLatestStepSummaries(executionId, ['b', 'c', 'ghost']); + + // one row per node asked about, the slim view: which slots were filled, not + // what they hold, and a node with no row is absent rather than null + expect(latest.b).toMatchObject({ + nodeId: 'b', + iteration: 1, + status: 'completed', + filledOutputSlots: [true, false], + }); + expect(latest.c).toMatchObject({ nodeId: 'c', iteration: 0 }); + expect(latest.ghost).toBeUndefined(); + + expect(await store.loadLatestStepSummaries(executionId, [])).toEqual({}); }); it('carries the unique key and the failed-rows partial index in the schema', async () => { diff --git a/packages/@n8n/engine/src/database/typeorm-step-store.ts b/packages/@n8n/engine/src/database/typeorm-step-store.ts index fad2975c342..408676528c3 100644 --- a/packages/@n8n/engine/src/database/typeorm-step-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-step-store.ts @@ -20,6 +20,26 @@ import { type StepSummary, } from '../execution/step-store'; +/** + * Per output slot: whether the step put data there. Computed in the query, so the + * potentially large outputs payload is never transferred. A slot counts as filled + * unless it holds JSON null. + */ +const FILLED_OUTPUT_SLOTS = `COALESCE( + (SELECT array_agg(jsonb_typeof(slot.value) <> 'null' ORDER BY slot.ordinality) + FROM jsonb_array_elements(step.outputs) WITH ORDINALITY AS slot), + '{}' +)`; + +/** What both summary queries select: every column but the outputs payload. */ +type StepSummaryRow = { + id: string; + nodeId: string; + iteration: number; + status: StepStatus; + filledOutputSlots: boolean[]; +}; + /** RETURNING rows come back keyed by database column name (snake_case). */ type InsertedStepRow = { id: string; node_id: string; iteration: number }; type ClaimedStepRow = { id: string; execution_id: string; node_id: string; iteration: number }; @@ -212,30 +232,14 @@ export class TypeOrmStepStore implements StepStore { ): Promise> { if (keys.length === 0) return {}; - // The per-slot booleans are computed inside the query, so the potentially - // large outputs payloads are never transferred. A slot counts as filled - // unless it holds JSON null. const { fragment, parameters } = stepKeyFilter(keys); - const rows: Array<{ - id: string; - nodeId: string; - iteration: number; - status: StepStatus; - filledOutputSlots: boolean[]; - }> = await this.repo + const rows: StepSummaryRow[] = await this.repo .createQueryBuilder('step') .select('step.id', 'id') .addSelect('step.node_id', 'nodeId') .addSelect('step.iteration', 'iteration') .addSelect('step.status', 'status') - .addSelect( - `COALESCE( - (SELECT array_agg(jsonb_typeof(slot.value) <> 'null' ORDER BY slot.ordinality) - FROM jsonb_array_elements(step.outputs) WITH ORDINALITY AS slot), - '{}' - )`, - 'filledOutputSlots', - ) + .addSelect(FILLED_OUTPUT_SLOTS, 'filledOutputSlots') .where('step.execution_id = :executionId', { executionId }) .andWhere(fragment, parameters) .getRawMany(); @@ -243,12 +247,27 @@ export class TypeOrmStepStore implements StepStore { return Object.fromEntries(rows.map((row) => [stepKeyId(row), row])); } - async loadLatestStep(executionId: string, nodeId: string): Promise { - // NOTE: `findOne({ where })`, not `findOneBy`, as in `loadStep`. - return await this.repo.findOne({ - where: { executionId, nodeId }, - order: { iteration: 'DESC' }, - }); + async loadLatestStepSummaries( + executionId: string, + nodeIds: string[], + ): Promise> { + if (nodeIds.length === 0) return {}; + + const rows: StepSummaryRow[] = await this.repo + .createQueryBuilder('step') + .distinctOn(['step.node_id']) + .select('step.id', 'id') + .addSelect('step.node_id', 'nodeId') + .addSelect('step.iteration', 'iteration') + .addSelect('step.status', 'status') + .addSelect(FILLED_OUTPUT_SLOTS, 'filledOutputSlots') + .where('step.execution_id = :executionId', { executionId }) + .andWhere('step.node_id IN (:...nodeIds)', { nodeIds }) + .orderBy('step.node_id') + .addOrderBy('step.iteration', 'DESC') + .getRawMany(); + + return Object.fromEntries(rows.map((row) => [row.nodeId, row])); } async countSettledSteps(executionId: string): Promise { diff --git a/packages/@n8n/engine/src/execution/__tests__/completion.test.ts b/packages/@n8n/engine/src/execution/__tests__/completion.test.ts new file mode 100644 index 00000000000..c320a3fe807 --- /dev/null +++ b/packages/@n8n/engine/src/execution/__tests__/completion.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it } from 'vitest'; + +import type { WorkflowLoop } from '../../graph'; +import { countExpectedSettledSteps } from '../completion'; + +function loop(batchNodeId: string, memberIds: string[]): WorkflowLoop { + return { + batchNodeId, + memberIds: new Set([batchNodeId, ...memberIds]), + backEdges: [], + entryEdges: [], + exitEdges: [], + }; +} + +describe('countExpectedSettledSteps', () => { + it('counts one row per node when there are no loops', () => { + expect(countExpectedSettledSteps([], new Set(['trigger', 'a', 'b']), new Map())).toBe(3); + }); + + it('counts a loop that ran three iterations', () => { + // B has rows 0, 1, 2 and 3, x has rows 0, 1 and 2, since the body has none + // at the terminal iteration. Plus trigger and d outside the loop. + const loops = [loop('B', ['x'])]; + const reachable = new Set(['trigger', 'B', 'x', 'd']); + + expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 3]]))).toBe(2 + 4 + 3); + }); + + it('counts a loop that ended immediately', () => { + // zero items: B's row 0 is terminal, and the body never ran + const loops = [loop('B', ['x'])]; + const reachable = new Set(['trigger', 'B', 'x', 'd']); + + expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 0]]))).toBe(2 + 1 + 0); + }); + + it('counts each member of a longer body per iteration', () => { + const loops = [loop('B', ['x', 'y', 'z'])]; + const reachable = new Set(['trigger', 'B', 'x', 'y', 'z']); + + // B: 0..2 is 3 rows, and x, y and z have 2 rows each + expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 2]]))).toBe(1 + 3 + 6); + }); + + it('owes an unknown number of rows while a loop is still running', () => { + const loops = [loop('B', ['x'])]; + const reachable = new Set(['trigger', 'B', 'x', 'd']); + + expect(countExpectedSettledSteps(loops, reachable, new Map())).toBeUndefined(); + }); + + it('ignores a loop the trigger cannot reach, which never receives rows', () => { + // the converter permits a loop component nothing points into: waiting on its + // ledger would hang an execution that is otherwise finished + const loops = [loop('B', ['x']), loop('orphan', ['other'])]; + const reachable = new Set(['trigger', 'B', 'x']); + + expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 1]]))).toBe(1 + 2 + 1); + }); + + it('waits on every reachable loop, not just the first', () => { + const loops = [loop('B1', ['x']), loop('B2', ['y'])]; + const reachable = new Set(['trigger', 'B1', 'x', 'B2', 'y']); + + expect(countExpectedSettledSteps(loops, reachable, new Map([['B1', 1]]))).toBeUndefined(); + expect( + countExpectedSettledSteps( + loops, + reachable, + new Map([ + ['B1', 1], + ['B2', 2], + ]), + ), + ).toBe(1 + (2 + 1) + (3 + 2)); + }); +}); diff --git a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts index 1b0fe553dc2..cea854aac09 100644 --- a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts @@ -31,7 +31,7 @@ function makeStepStore(createSteps = vi.fn()): StepStore { cancelQueuedSteps: vi.fn(), loadStepsByKeys: vi.fn().mockResolvedValue({}), loadStepSummariesByKeys: vi.fn().mockResolvedValue({}), - loadLatestStep: vi.fn().mockResolvedValue(null), + loadLatestStepSummaries: vi.fn().mockResolvedValue({}), countSettledSteps: vi.fn(), hasFailedSteps: vi.fn(), }; diff --git a/packages/@n8n/engine/src/execution/__tests__/loop-ledger.test.ts b/packages/@n8n/engine/src/execution/__tests__/loop-ledger.test.ts new file mode 100644 index 00000000000..f9f3946a61e --- /dev/null +++ b/packages/@n8n/engine/src/execution/__tests__/loop-ledger.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { deriveLoops, type WorkflowGraph } from '../../graph'; +import { endsLoop, exitSourcesInto, isTerminalStep, loadTerminalIterations } from '../loop-ledger'; +import type { StepStore, StepSummary } from '../step-store'; + +function tip(iteration: number, filledOutputSlots: boolean[], status = 'completed' as const) { + return { id: `step-B-${iteration}`, nodeId: 'B', iteration, status, filledOutputSlots }; +} + +/** The canonical loop: trigger into B, body x, exit to d. */ +const loopGraph: WorkflowGraph = { + nodes: [ + { id: 'trigger', name: 'T', type: 'trigger' }, + { id: 'B', name: 'B', type: 'batch' }, + { id: 'x', name: 'X', type: 'v1-node' }, + { id: 'd', name: 'D', type: 'v1-node' }, + ], + edges: [ + { from: 'trigger', to: 'B', outputIndex: 0, inputIndex: 0 }, + { from: 'B', to: 'x', outputIndex: 1, inputIndex: 0 }, + { from: 'x', to: 'B', outputIndex: 0, inputIndex: 0, isBackEdge: true }, + { from: 'B', to: 'd', outputIndex: 0, inputIndex: 0 }, + ], +}; +const loops = deriveLoops(loopGraph); + +describe('endsLoop', () => { + it('ends the loop when a settled row leaves its loop slot dead', () => { + expect(endsLoop('completed', false)).toBe(true); + }); + + it('does not end the loop while the loop slot is filled', () => { + expect(endsLoop('completed', true)).toBe(false); + }); + + it('ends the loop on a skipped row, which fired nothing', () => { + expect(endsLoop('skipped', false)).toBe(true); + }); + + it('does not end the loop from a row that has not settled', () => { + expect(endsLoop('running', false)).toBe(false); + expect(endsLoop('queued', false)).toBe(false); + }); +}); + +describe('isTerminalStep', () => { + it('is terminal when the row fired the done slot', () => { + expect(isTerminalStep(tip(2, [true, false]))).toBe(true); + }); + + it('is not terminal when the row fired the loop slot', () => { + expect(isTerminalStep(tip(2, [false, true]))).toBe(false); + }); + + it('is terminal when the row fired nothing at all', () => { + expect(isTerminalStep(tip(2, []))).toBe(true); + }); +}); + +describe('exitSourcesInto', () => { + it('names the batch node whose terminal row a node after the loop reads', () => { + expect(exitSourcesInto(loopGraph, loops, ['d'])).toEqual(['B']); + }); + + it('names nothing for a node inside the loop, which reads its own iteration', () => { + expect(exitSourcesInto(loopGraph, loops, ['x'])).toEqual([]); + expect(exitSourcesInto(loopGraph, loops, ['B'])).toEqual([]); + }); + + it('names nothing in a graph without loops, so no tip is ever read', () => { + expect(exitSourcesInto(loopGraph, [], ['d'])).toEqual([]); + }); +}); + +describe('loadTerminalIterations', () => { + function makeStepStore(latest: Record): StepStore { + return { loadLatestStepSummaries: vi.fn().mockResolvedValue(latest) } as unknown as StepStore; + } + + it('reports the terminal iteration of a loop that has ended', async () => { + const stepStore = makeStepStore({ B: tip(3, [true, false]) }); + + expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map([['B', 3]])); + expect(stepStore.loadLatestStepSummaries).toHaveBeenCalledExactlyOnceWith('exec-1', ['B']); + }); + + it('omits a loop still running, so its exit stays undecidable', async () => { + const stepStore = makeStepStore({ B: tip(3, [false, true]) }); + + expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map()); + }); + + it('omits a loop with no rows yet', async () => { + const stepStore = makeStepStore({}); + + expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map()); + }); + + it('reads several loops in one query, ended and running alike', async () => { + const stepStore = makeStepStore({ + B1: { ...tip(2, [true, false]), nodeId: 'B1' }, + B2: { ...tip(5, [false, true]), nodeId: 'B2' }, + }); + + expect(await loadTerminalIterations(stepStore, 'exec-1', ['B1', 'B2'])).toEqual( + new Map([['B1', 2]]), + ); + expect(stepStore.loadLatestStepSummaries).toHaveBeenCalledExactlyOnceWith('exec-1', [ + 'B1', + 'B2', + ]); + }); + + it('reads nothing when no loop is named', async () => { + const stepStore = makeStepStore({}); + + expect(await loadTerminalIterations(stepStore, 'exec-1', [])).toEqual(new Map()); + expect(stepStore.loadLatestStepSummaries).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/@n8n/engine/src/execution/__tests__/settlement.test.ts b/packages/@n8n/engine/src/execution/__tests__/settlement.test.ts index ccc20119efc..17598da7cd5 100644 --- a/packages/@n8n/engine/src/execution/__tests__/settlement.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/settlement.test.ts @@ -1,9 +1,9 @@ import { describe, expect, it } from 'vitest'; import type { GraphEdge, WorkflowGraph } from '../../graph'; -import { getDescendantNodeIds } from '../../graph'; +import { deriveLoops, getDescendantNodeIds } from '../../graph'; import { stepKeyId, type StepKey, type StepStatus } from '../execution.types'; -import { decideSuccessors } from '../settlement'; +import { decideSuccessors, type SuccessorDecisions } from '../settlement'; import type { StepSummary } from '../step-store'; function summary( @@ -18,11 +18,19 @@ function makeSteps(...summaries: StepSummary[]): Record { return Object.fromEntries(summaries.map((s) => [stepKeyId(s), s])); } -/** Every row in these tests sits at iteration 0; loops are CAT-2875 part 2. */ function keyFor(nodeId: string): StepKey { return { nodeId, iteration: 0 }; } +/** No loops, so there is no loop set and no loop that could have ended. */ +function decideLoopless( + graph: WorkflowGraph, + settled: StepKey, + steps: Record, +): SuccessorDecisions { + return decideSuccessors(graph, [], settled, steps, new Map()); +} + function makeGraph( edges: Array & Pick>, ): WorkflowGraph { @@ -50,7 +58,7 @@ describe('decideSuccessors', () => { it('queues the live branch and skips the dead one', () => { // if fired only output slot 0; m is not a direct successor, so it is // not considered — b's own settled event will examine it. - const decisions = decideSuccessors( + const decisions = decideLoopless( diamond, keyFor('if'), makeSteps(summary('trigger', 'completed', [true]), summary('if', 'completed', [true, false])), @@ -62,7 +70,7 @@ describe('decideSuccessors', () => { it('decides a merge once its last predecessor settles', () => { // b was skipped earlier; a just completed. m has one live and one dead // edge, so it runs on the live data. - const decisions = decideSuccessors( + const decisions = decideLoopless( diamond, keyFor('a'), makeSteps( @@ -79,7 +87,7 @@ describe('decideSuccessors', () => { it('leaves a merge undecided while a predecessor is still unsettled', () => { // b's settled event fires while a is still queued: m is not decidable // yet, and a's own settlement will decide it later. - const decisions = decideSuccessors( + const decisions = decideLoopless( diamond, keyFor('b'), makeSteps( @@ -107,19 +115,19 @@ describe('decideSuccessors', () => { summary('if', 'completed', [true, false]), ); - expect(decideSuccessors(graph, keyFor('if'), steps)).toEqual({ + expect(decideLoopless(graph, keyFor('if'), steps)).toEqual({ toQueue: [keyFor('a')], toSkip: [keyFor('b')], }); steps[stepKeyId(keyFor('b'))] = summary('b', 'skipped'); - expect(decideSuccessors(graph, keyFor('b'), steps)).toEqual({ + expect(decideLoopless(graph, keyFor('b'), steps)).toEqual({ toQueue: [], toSkip: [keyFor('c')], }); steps[stepKeyId(keyFor('c'))] = summary('c', 'skipped'); - expect(decideSuccessors(graph, keyFor('c'), steps)).toEqual({ + expect(decideLoopless(graph, keyFor('c'), steps)).toEqual({ toQueue: [], toSkip: [keyFor('d')], }); @@ -142,10 +150,10 @@ describe('decideSuccessors', () => { summary('b', 'skipped'), ); - expect(decideSuccessors(graph, keyFor('b'), steps)).toEqual({ toQueue: [], toSkip: [] }); + expect(decideLoopless(graph, keyFor('b'), steps)).toEqual({ toQueue: [], toSkip: [] }); steps[stepKeyId(keyFor('c'))] = summary('c', 'skipped'); - expect(decideSuccessors(graph, keyFor('c'), steps)).toEqual({ + expect(decideLoopless(graph, keyFor('c'), steps)).toEqual({ toQueue: [], toSkip: [keyFor('d')], }); @@ -155,7 +163,7 @@ describe('decideSuccessors', () => { // duplicate delivery: both successors were planned by the first run. // Nothing is created, so nothing is announced — a lost announcement is // reconciliation's job (CAT-2938), not this planner's. - const decisions = decideSuccessors( + const decisions = decideLoopless( diamond, keyFor('if'), makeSteps( @@ -177,7 +185,7 @@ describe('decideSuccessors', () => { { from: 'switch', to: 'c', outputIndex: 2 }, ]); - const decisions = decideSuccessors( + const decisions = decideLoopless( graph, keyFor('switch'), makeSteps( @@ -195,7 +203,7 @@ describe('decideSuccessors', () => { { from: 'a', to: 'b', outputIndex: 1 }, ]); - const decisions = decideSuccessors( + const decisions = decideLoopless( graph, keyFor('a'), makeSteps(summary('trigger', 'completed', [true]), summary('a', 'completed', [true])), @@ -274,7 +282,7 @@ function simulateExecution( if (handleEvent) { const nodeId = settledEvents[0]; settledEvents.shift(); - const { toQueue, toSkip } = decideSuccessors(graph, keyFor(nodeId), steps); + const { toQueue, toSkip } = decideLoopless(graph, keyFor(nodeId), steps); for (const { nodeId: id } of toQueue) { steps[stepKeyId(keyFor(id))] = summary(id, 'queued'); queuedNodes.push(id); @@ -336,3 +344,170 @@ describe('the event loop over decideSuccessors matches the reference evaluator', } }); }); + +describe('decideSuccessors over loop iterations', () => { + /** + * ┌───────┐ ┌───┐ o0 ┌───┐ + * │trigger├───►│ ├──────►│ d │ + * └───────┘ │ B │ └───┘ + * │ │ o1 ┌───┐ + * │ ├──────►│ x │ + * └─▲─┘ └─┬─┘ + * └──(back)───┘ + */ + const graph = makeGraph([ + { from: 'trigger', to: 'B' }, + { from: 'B', to: 'x', outputIndex: 1 }, + { from: 'x', to: 'B', isBackEdge: true }, + { from: 'B', to: 'd', outputIndex: 0 }, + ]); + graph.nodes = graph.nodes.map((node) => + node.id === 'B' ? { ...node, type: 'batch' as const } : node, + ); + const loops = deriveLoops(graph); + + /** A row at a given iteration, where the loopless `summary` always builds 0. */ + function at( + nodeId: string, + iteration: number, + status: StepStatus, + filledOutputSlots: boolean[] = [], + ): StepSummary { + return { id: `step-${nodeId}-${iteration}`, nodeId, iteration, status, filledOutputSlots }; + } + + function decide( + settled: StepKey, + steps: Record, + terminalIterations: Map = new Map(), + ): SuccessorDecisions { + return decideSuccessors(graph, loops, settled, steps, terminalIterations); + } + + const key = (nodeId: string, iteration: number): StepKey => ({ nodeId, iteration }); + + it('queues the body at the batch row iteration, and leaves the exit undecided', () => { + // B filled its loop slot, so the loop runs on and its end is unknown + const steps = makeSteps(at('B', 0, 'completed', [false, true])); + + expect(decide(key('B', 0), steps)).toEqual({ toQueue: [key('x', 0)], toSkip: [] }); + }); + + it('advances the iteration across the return edge', () => { + const steps = makeSteps( + at('B', 0, 'completed', [false, true]), + at('x', 0, 'completed', [true]), + ); + + expect(decide(key('x', 0), steps)).toEqual({ toQueue: [key('B', 1)], toSkip: [] }); + }); + + it('skips the next iteration when the body returns nothing', () => { + // a dead return edge ends the loop: (B, 1) is skipped at birth, and skipped + // rows fire nothing, which makes it the terminal row + const steps = makeSteps( + at('B', 0, 'completed', [false, true]), + at('x', 0, 'completed', [false]), + ); + + expect(decide(key('x', 0), steps)).toEqual({ toQueue: [], toSkip: [key('B', 1)] }); + }); + + it('queues what follows the loop from the terminal row only', () => { + // B filled its done slot instead, so the loop has ended at iteration 2 + const steps = makeSteps(at('B', 2, 'completed', [true, false])); + const terminals = new Map([['B', 2]]); + + expect(decide(key('B', 2), steps, terminals)).toEqual({ + toQueue: [key('d', 0)], + toSkip: [], + }); + }); + + it('plans no body row at the terminal iteration', () => { + // the cascade this prevents: a skipped body row would settle, its return edge + // would plan another batch row, and that row would end the loop in turn + const steps = makeSteps(at('B', 2, 'completed', [true, false])); + const terminals = new Map([['B', 2]]); + + const decisions = decide(key('B', 2), steps, terminals); + + expect(decisions.toQueue).not.toContainEqual(key('x', 2)); + expect(decisions.toSkip).not.toContainEqual(key('x', 2)); + }); + + it('leaves what follows the loop undecided while the loop still runs', () => { + // B@1 has a dead done slot, and d is not skipped on the strength of it: an + // empty toSkip is the assertion, since a later row may still fire that slot + const steps = makeSteps( + at('B', 0, 'completed', [false, true]), + at('x', 0, 'completed', [true]), + at('B', 1, 'completed', [false, true]), + ); + + expect(decide(key('B', 1), steps)).toEqual({ toQueue: [key('x', 1)], toSkip: [] }); + }); + + /** + * The rule above keeps a running loop from deciding its exit, but a node after + * the loop can also be reached from outside it. Then the exit edge is resolved + * for real, finds no terminal row, and holds the decision open. + * + * ┌───────┐ ┌───┐ o1 ┌───┐ + * │trigger├───►│ B ├──────►│ x │ + * └───┬───┘ └─▲─┘ └─┬─┘ + * │ └──(back)───┘ + * │ ┌───┐ o0 + * └──────►│ p ├──────────► d ◄── B's done slot + * └───┘ + */ + it('holds a node after the loop undecided when another predecessor settles first', () => { + const joined = makeGraph([ + { from: 'trigger', to: 'B' }, + { from: 'B', to: 'x', outputIndex: 1 }, + { from: 'x', to: 'B', isBackEdge: true }, + { from: 'B', to: 'd', outputIndex: 0 }, + { from: 'trigger', to: 'p', outputIndex: 1 }, + { from: 'p', to: 'd', inputIndex: 1 }, + ]); + joined.nodes = joined.nodes.map((node) => + node.id === 'B' ? { ...node, type: 'batch' as const } : node, + ); + const joinedLoops = deriveLoops(joined); + const steps = makeSteps( + at('B', 0, 'completed', [false, true]), + at('p', 0, 'completed', [true]), + ); + + // p settling makes d a candidate, but the loop has not ended, so the exit + // edge has no row to read and d gets no fate at all + expect(decideSuccessors(joined, joinedLoops, key('p', 0), steps, new Map())).toEqual({ + toQueue: [], + toSkip: [], + }); + + // once it has ended, the same settlement queues d + const ended = makeSteps( + at('B', 2, 'completed', [true, false]), + at('p', 0, 'completed', [true]), + ); + expect(decideSuccessors(joined, joinedLoops, key('p', 0), ended, new Map([['B', 2]]))).toEqual({ + toQueue: [key('d', 0)], + toSkip: [], + }); + }); + + it('reads the entry edge at iteration 0 and the return edge after it', () => { + const entry = makeSteps(at('trigger', 0, 'completed', [true])); + expect(decide(key('trigger', 0), entry)).toEqual({ toQueue: [key('B', 0)], toSkip: [] }); + + // at iteration 1 the entry edge connects nothing, so the return edge alone + // decides, and B is queued on its strength + const second = makeSteps( + at('trigger', 0, 'completed', [true]), + at('B', 0, 'completed', [false, true]), + at('x', 0, 'completed', [true]), + ); + expect(decide(key('x', 0), second)).toEqual({ toQueue: [key('B', 1)], toSkip: [] }); + }); +}); 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 3c61e361059..f735c64524a 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 @@ -1,12 +1,12 @@ import { describe, expect, it, vi } from 'vitest'; import type { ExternalDependencies, IStepExecutor } from '../../dependencies'; -import type { WorkflowGraph } from '../../graph'; +import { deriveLoops, type WorkflowGraph } from '../../graph'; import type { OrchestrationMessage, WorkQueue } from '../../queue'; import type { ExecutionRecord, ExecutionStore } from '../execution-store'; import { stepKeyId, type StepSlots, type StepStatus } from '../execution.types'; -import { StepReadyHandler } from '../step-ready-handler'; -import type { StepRecord, StepStore } from '../step-store'; +import { resolveInputReads, StepReadyHandler } from '../step-ready-handler'; +import type { StepRecord, StepStore, StepSummary } from '../step-store'; /** Key for a `loadStepsByKeys` result at iteration 0, as the handler requests them. */ const at = (nodeId: string) => stepKeyId({ nodeId, iteration: 0 }); @@ -77,7 +77,7 @@ function makeStepStore(step: Partial = {}, overrides: Partial { }, ); }); + +describe('StepReadyHandler over loop iterations', () => { + /** + * ┌───────┐ ┌───┐ o0 ┌───┐ + * │trigger├───►│ ├──────►│ d │ + * └───────┘ │ B │ └───┘ + * │ │ o1 ┌───┐ + * │ ├──────►│ x │ + * └─▲─┘ └─┬─┘ + * └──(back)───┘ + */ + const loopGraph: WorkflowGraph = { + nodes: [ + { id: 'trigger', name: 'T', type: 'trigger' }, + { id: 'B', name: 'B', type: 'batch' }, + { id: 'x', name: 'X', type: 'v1-node' }, + { id: 'd', name: 'D', type: 'v1-node' }, + ], + edges: [ + { from: 'trigger', to: 'B', outputIndex: 0, inputIndex: 0 }, + { from: 'B', to: 'x', outputIndex: 1, inputIndex: 0 }, + { from: 'x', to: 'B', outputIndex: 0, inputIndex: 0, isBackEdge: true }, + { from: 'B', to: 'd', outputIndex: 0, inputIndex: 0 }, + ], + }; + + /** A loop's latest row as the store returns it: filled slots, no payloads. */ + function tipAt(iteration: number, filledOutputSlots: boolean[]): StepSummary { + return { + id: `step-B-${iteration}`, + nodeId: 'B', + iteration, + status: 'completed', + filledOutputSlots, + }; + } + + function rowAt(nodeId: string, iteration: number, outputs: StepRecord['outputs']): StepRecord { + return { + id: `step-${nodeId}-${iteration}`, + executionId: 'exec-1', + nodeId, + iteration, + status: 'completed', + outputs, + error: null, + }; + } + + it('reads the body input from the batch row at the same iteration', async () => { + const executor = makeExecutor(); + const stepStore = makeStepStore( + { id: 'step-x-2', nodeId: 'x', iteration: 2 }, + { + loadStepsByKeys: vi.fn().mockResolvedValue({ + [stepKeyId({ nodeId: 'B', iteration: 2 })]: rowAt('B', 2, [null, [{ json: { i: 2 } }]]), + }), + }, + ); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: loopGraph }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await handler.handle({ ...event, stepId: 'step-x-2' }); + + expect(stepStore.loadStepsByKeys).toHaveBeenCalledWith('exec-1', [ + { nodeId: 'B', iteration: 2 }, + ]); + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ inputs: [[{ json: { i: 2 } }]] }), + ); + }); + + /** + * A batch node's slot 0 carries both its entry edge and its return edge, which + * the old check rejected outright as two edges into one slot. Asserted on the + * resolution rather than through the handler, since running a batch step needs + * an executor that does not exist yet. + */ + it('reads the batch node from the entry edge at iteration 0 and the return edge after it', () => { + const loops = deriveLoops(loopGraph); + const intoB = loopGraph.edges.filter((edge) => edge.to === 'B'); + + const readsAt = (iteration: number) => + resolveInputReads( + intoB, + loops, + { id: `step-B-${iteration}`, nodeId: 'B', iteration }, + new Map(), + ).map(({ edge, key }) => ({ from: edge.from, key })); + + expect(readsAt(0)).toEqual([{ from: 'trigger', key: { nodeId: 'trigger', iteration: 0 } }]); + expect(readsAt(1)).toEqual([{ from: 'x', key: { nodeId: 'x', iteration: 0 } }]); + }); + + it('reads what follows the loop from the terminal row, whatever iteration that is', async () => { + const executor = makeExecutor(); + const stepStore = makeStepStore( + { id: 'step-d-0', nodeId: 'd', iteration: 0 }, + { + loadLatestStepSummaries: vi.fn().mockResolvedValue({ B: tipAt(4, [true, false]) }), + loadStepsByKeys: vi.fn().mockResolvedValue({ + [stepKeyId({ nodeId: 'B', iteration: 4 })]: rowAt('B', 4, [ + [{ json: { done: true } }], + null, + ]), + }), + }, + ); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: loopGraph }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await handler.handle({ ...event, stepId: 'step-d-0' }); + + expect(stepStore.loadStepsByKeys).toHaveBeenCalledWith('exec-1', [ + { nodeId: 'B', iteration: 4 }, + ]); + expect(executor.execute).toHaveBeenCalledWith( + expect.objectContaining({ inputs: [[{ json: { done: true } }]] }), + ); + }); + + it('throws, running nothing, when the loop it reads across has not ended', async () => { + // only the planner should queue such a step, so the rows and the plan disagree + const executor = makeExecutor(); + const stepStore = makeStepStore( + { id: 'step-d-0', nodeId: 'd', iteration: 0 }, + { loadLatestStepSummaries: vi.fn().mockResolvedValue({ B: tipAt(4, [false, true]) }) }, + ); + const handler = new StepReadyHandler( + makeExecutionStore({ graph: loopGraph }), + stepStore, + makeQueue(), + { v1StepExecutor: executor }, + ); + + await expect(handler.handle({ ...event, stepId: 'step-d-0' })).rejects.toThrow( + /across a loop that has not ended/, + ); + expect(executor.execute).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts index 60d64c314ac..b777cff6b06 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts @@ -103,7 +103,7 @@ function makeStepStore( ); }), loadStepsByKeys: vi.fn().mockResolvedValue({}), - loadLatestStep: vi.fn().mockResolvedValue(null), + loadLatestStepSummaries: vi.fn().mockResolvedValue({}), // far from settled, so finish tests opt in explicitly countSettledSteps: vi.fn().mockResolvedValue(0), hasFailedSteps: vi.fn().mockResolvedValue(false), @@ -201,7 +201,7 @@ describe('StepSettledHandler', () => { expect(orchestrationQueue.publish).not.toHaveBeenCalled(); }); - it('loads the decision rows in one query: the successors and their predecessors', async () => { + it('loads the decision rows in one query: the settled row, the successors, and what they read', async () => { const stepStore = makeStepStore({ id: 'step-b', nodeId: 'b' }, {}, [ ...defaultSummaries, summary('b', 'completed', [true]), @@ -210,10 +210,10 @@ describe('StepSettledHandler', () => { await handler.handle({ ...event, stepId: 'step-b' }); - // b's only successor is m; m reads from both b and c + // b's only successor is m, and m reads from both b and c expect(stepStore.loadStepSummariesByKeys).toHaveBeenCalledExactlyOnceWith('exec-1', [ - { nodeId: 'm', iteration: 0 }, { nodeId: 'b', iteration: 0 }, + { nodeId: 'm', iteration: 0 }, { nodeId: 'c', iteration: 0 }, ]); }); diff --git a/packages/@n8n/engine/src/execution/completion.ts b/packages/@n8n/engine/src/execution/completion.ts new file mode 100644 index 00000000000..8a1d2a1b42c --- /dev/null +++ b/packages/@n8n/engine/src/execution/completion.ts @@ -0,0 +1,42 @@ +import type { WorkflowLoop } from '../graph'; + +/** + * How many settled steps a finished execution owes, or `undefined` while a loop + * the trigger reaches is still running. + * + * Without loops this is the reachable node count, one step per node. A loop + * replaces its members' single steps with one per iteration. + * + * While a loop runs, how many steps it will create is not known, so there is no + * total to give. That matters between two iterations, when every step so far has + * settled: with a total, the execution would look finished when the next + * iteration simply has not been planned yet. + * + * Only loops the trigger reaches count. The converter allows a loop component + * nothing points into, and those never receive steps, so waiting on one would + * hang the execution. + */ +export function countExpectedSettledSteps( + loops: WorkflowLoop[], + reachable: Set, + terminalIterations: Map, +): number | undefined { + let expected = 0; + const members = new Set(); + + for (const loop of loops.filter((l) => reachable.has(l.batchNodeId))) { + const terminal = terminalIterations.get(loop.batchNodeId); + if (terminal === undefined) return undefined; + + // iterations 0 to t for the batch node, then t each for the rest: the body + // has no step at the terminal iteration + expected += terminal + 1 + terminal * (loop.memberIds.size - 1); + for (const memberId of loop.memberIds) members.add(memberId); + } + + for (const nodeId of reachable) { + if (!members.has(nodeId)) expected += 1; + } + + return expected; +} diff --git a/packages/@n8n/engine/src/execution/loop-ledger.ts b/packages/@n8n/engine/src/execution/loop-ledger.ts new file mode 100644 index 00000000000..55caf705fc8 --- /dev/null +++ b/packages/@n8n/engine/src/execution/loop-ledger.ts @@ -0,0 +1,72 @@ +import type { WorkflowGraph, WorkflowLoop } from '../graph'; +import { isSettledStatus, type StepStatus } from './execution.types'; +import { classifyEdge } from './iteration-mapping'; +import type { StepStore, StepSummary } from './step-store'; + +/** A batch node's output slots: 0 is done, 1 is loop. */ +const LOOP_SLOT = 1; + +/** + * A loop's batch node steps form a ledger: one per iteration, written strictly + * in order, since iteration `i + 1` is planned only once iteration `i` has + * settled. The last one is the terminal step, and it is what says the loop is + * over. + * + * A step ends the loop when it settles without filling its loop slot: it fired + * the done slot instead, or it never ran at all, as a skip records. Nothing can + * advance the loop past it, so it is always the last one. + */ +export function endsLoop(status: StepStatus, loopSlotFilled: boolean): boolean { + return isSettledStatus(status) && !loopSlotFilled; +} + +/** `endsLoop` for a step, which carries both of the facts it asks for. */ +export function isTerminalStep(step: StepSummary): boolean { + return endsLoop(step.status, Boolean(step.filledOutputSlots[LOOP_SLOT])); +} + +/** + * The batch nodes whose loop-ending step has to be read to resolve the edges + * into `targetNodeIds`. Only an exit edge reads one, so a target with no exit + * edge into it needs none: every other edge class resolves against the target's + * own iteration. + */ +export function exitSourcesInto( + graph: WorkflowGraph, + loops: WorkflowLoop[], + targetNodeIds: string[], +): string[] { + if (loops.length === 0) return []; + + const targets = new Set(targetNodeIds); + const sources = new Set(); + for (const edge of graph.edges) { + if (targets.has(edge.to) && classifyEdge(edge, loops) === 'exit') sources.add(edge.from); + } + return [...sources]; +} + +/** + * The terminal iteration of each loop asked about, by batch node id, omitting + * the loops that have not ended. + * + * Only the latest step can be terminal, since they are written in order, so one + * query over every loop answers this. It reads the slim view because the step + * ending a loop holds everything that loop accumulated, and this runs on every + * settlement. + */ +export async function loadTerminalIterations( + stepStore: StepStore, + executionId: string, + batchNodeIds: string[], +): Promise> { + if (batchNodeIds.length === 0) return new Map(); + + const latest = await stepStore.loadLatestStepSummaries(executionId, batchNodeIds); + + const terminalIterations = new Map(); + for (const [batchNodeId, step] of Object.entries(latest)) { + if (isTerminalStep(step)) terminalIterations.set(batchNodeId, step.iteration); + } + return terminalIterations; +} diff --git a/packages/@n8n/engine/src/execution/settlement.ts b/packages/@n8n/engine/src/execution/settlement.ts index 2134e61e13e..1cfcfb754b8 100644 --- a/packages/@n8n/engine/src/execution/settlement.ts +++ b/packages/@n8n/engine/src/execution/settlement.ts @@ -1,90 +1,214 @@ -import { getSuccessorNodeIds, type GraphEdge, type WorkflowGraph } from '../graph'; +import type { GraphEdge, WorkflowGraph, WorkflowLoop } from '../graph'; import { stepKeyId, isSettledStatus, type StepKey, type StepKeyId } from './execution.types'; +import { classifyEdge, sourceRow, targetKey, type EdgeClass } from './iteration-mapping'; +import { isTerminalStep } from './loop-ledger'; import type { StepSummary } from './step-store'; /** + * A **step** is one run of one node, identified by `StepKey`, which is + * `(nodeId, iteration)`. Outside a loop a node has a single step, at iteration + * 0. A loop member has one step per pass. + * * Settlement rules (design: CAT-2874): * - * 1. Every node the execution considers eventually settles: completed, - * failed, skipped, or cancelled. Settled fates are immutable. The - * execution is finished exactly when every reachable node has settled. + * 1. Every step the execution creates eventually settles: completed, failed, + * skipped, or cancelled. Settled fates are immutable. The execution is + * finished exactly when every step it owes has settled, which + * `completion.ts` counts. * 2. An edge is live iff its source completed and filled the edge's output * slot; it is dead if the source settled any other way, or left the slot * null. An edge whose source is unsettled is neither yet. - * 3. A node is decidable once every predecessor is settled. Decidable with at - * least one live incoming edge -> it runs (queued). Decidable with none -> - * it is skipped: settled at birth, its own out-edges all dead. + * 3. A step is decidable once every step its incoming edges read has settled. + * Decidable with at least one live incoming edge -> it runs (queued). + * Decidable with none -> it is skipped: settled at birth, its own out-edges + * all dead. * 4. A settlement decides only its direct successors; the cascade is the * event loop. A skip is announced as settled like any other settlement, * and its handler decides the next hop. * 5. A planner commits its decisions in one batch, then announces only the - * rows it created. A settlement whose announcement is lost is detectable - * from rows alone — a settled row with decidable but rowless successors — - * and re-announced by reconciliation (CAT-2938). + * steps it created. A settlement whose announcement is lost is detectable + * from the steps alone — a settled one whose successors are decidable but + * absent — and re-announced by reconciliation (CAT-2938). * - * Fates are pure functions of settled rows, so any planner, at any time, + * Fates are pure functions of settled steps, so any planner, at any time, * recomputes the same decisions; duplicates and races converge instead of * corrupting. * - * Rows are identified per `(nodeId, iteration)`: loop members run once per - * pass (CAT-2875). Forward edges stay within a pass; the back-edge and exit - * mappings land with loop execution, and until then back-edges are rejected. + * Loops extend those rules rather than replacing them (CAT-2875). + * `iteration-mapping.ts` decides which iterations an edge connects, and two + * consequences belong to loops alone: + * + * - A batch node's step decides one side of its loop. While the loop runs it + * decides the body, and on the terminal step, the one whose loop slot stayed + * dead, it decides the nodes after the loop instead. Deciding the other side + * too early would skip it, and a later step could not take that back. + * - Body steps exist for running iterations only. The terminal iteration has + * none at all, not even skipped ones, or those skips would cascade through + * the body into a further iteration, and on forever. The exclusion lives in + * `decideNodeFate`, so anything recomputing fates reaches the same answer, + * reconciliation included. */ export interface SuccessorDecisions { - /** Successors with a live input, to enqueue — in edge order. */ + /** Successor steps with a live input, to enqueue — in edge order. */ toQueue: StepKey[]; - /** Successors with settled but all-dead inputs, to record as skipped. */ + /** Successor steps with settled but all-dead inputs, to record as skipped. */ toSkip: StepKey[]; } /** - * Decides the direct successors of the settled step (rules 2–4). `steps` - * holds the existing rows for those successors and their predecessors, keyed - * by `stepKeyId`, including the settled step itself. + * Decides the direct successors of the settled step (rules 2–4). + * + * `steps` holds the steps `decisionKeys` names, keyed by `stepKeyId`. Anything + * less and a decision reads an absent step as one that has not settled, leaving + * the successor undecided forever. `terminalIterations` holds each loop's + * terminal iteration by batch node id, omitting the loops that have not ended. */ export function decideSuccessors( graph: WorkflowGraph, + loops: WorkflowLoop[], settled: StepKey, steps: Record, + terminalIterations: Map, ): SuccessorDecisions { const decisions: SuccessorDecisions = { toQueue: [], toSkip: [] }; - for (const successorNodeId of getSuccessorNodeIds(graph, settled.nodeId)) { - const successor: StepKey = { nodeId: successorNodeId, iteration: settled.iteration }; - // An existing row was decided by an earlier settlement, which announced it. - if (steps[stepKeyId(successor)]) continue; - const fate = decideNodeFate(graph, successor, steps); - if (fate === 'queued') decisions.toQueue.push(successor); - else if (fate === 'skipped') decisions.toSkip.push(successor); + const batchStep = loops.some((loop) => loop.batchNodeId === settled.nodeId) + ? steps[stepKeyId(settled)] + : undefined; + const decided = new Set(); + const outgoingEdges = graph.edges.filter((edge) => edge.from === settled.nodeId); + + for (const edge of outgoingEdges) { + const edgeClass = classifyEdge(edge, loops); + if (batchStep && !batchStepDecides(edgeClass, batchStep)) continue; + + const target = targetKey(edge, edgeClass, settled); + const targetId = stepKeyId(target); + // An existing step was decided by an earlier settlement, which announced it. + // Two edges into one step are one candidate, decided once. + if (steps[targetId] || decided.has(targetId)) continue; + decided.add(targetId); + + const fate = decideNodeFate(graph, loops, target, steps, terminalIterations); + if (fate === 'queued') decisions.toQueue.push(target); + else if (fate === 'skipped') decisions.toSkip.push(target); } + return decisions; } -/** One candidate's fate under rules 2–3; undecidable while a predecessor is unsettled. */ +/** Which side of its loop a batch step decides: the body, or what follows. */ +function batchStepDecides(edgeClass: EdgeClass, batchStep: StepSummary): boolean { + return isTerminalStep(batchStep) ? edgeClass === 'exit' : edgeClass !== 'exit'; +} + +/** + * One candidate's fate under rules 2–3. + * + * `undecidable` covers a predecessor that has not settled and a loop that has + * not ended alike, since either way the step an edge reads does not exist yet. + * `outside` means the candidate is not part of the execution at this iteration + * and gets no step at all, which is what keeps a loop that has ended from + * cascading skips through its own body. + */ function decideNodeFate( graph: WorkflowGraph, + loops: WorkflowLoop[], candidate: StepKey, steps: Record, -): 'queued' | 'skipped' | 'undecidable' { - // Back-edges aside: loop iteration is CAT-2875. - const incoming = graph.edges.filter((edge) => edge.to === candidate.nodeId && !edge.isBackEdge); - const predecessors = [...new Set(incoming.map((edge) => edge.from))]; - const settled = predecessors.every((nodeId) => { - const row = steps[stepKeyId({ nodeId, iteration: candidate.iteration })]; - return row !== undefined && isSettledStatus(row.status); - }); - if (!settled) return 'undecidable'; - return incoming.some((edge) => isLiveEdge(edge, candidate.iteration, steps)) - ? 'queued' - : 'skipped'; + terminalIterations: Map, +): 'queued' | 'skipped' | 'undecidable' | 'outside' { + if (isPastLoopEnd(loops, candidate, steps)) return 'outside'; + + let applicable = 0; + let live = false; + const incomingEdges = graph.edges.filter((edge) => edge.to === candidate.nodeId); + + for (const edge of incomingEdges) { + const source = sourceRow( + edge, + classifyEdge(edge, loops), + candidate, + terminalIterations.get(edge.from), + ); + if (source.kind === 'none') continue; + if (source.kind === 'pending') return 'undecidable'; + + applicable += 1; + const sourceStep = steps[stepKeyId(source.key)]; + if (!sourceStep || !isSettledStatus(sourceStep.status)) return 'undecidable'; + if (isLive(sourceStep, edge)) live = true; + } + + // No edge connects to this step at this iteration, so nothing produces it. + if (applicable === 0) return 'outside'; + + return live ? 'queued' : 'skipped'; +} + +/** + * Whether the candidate is a body step of a loop that has already ended. Read + * from the batch step rather than from `terminalIterations`, so the exclusion + * holds for anything recomputing fates from the steps alone. + */ +function isPastLoopEnd( + loops: WorkflowLoop[], + candidate: StepKey, + steps: Record, +): boolean { + const loop = loops.find((l) => l.memberIds.has(candidate.nodeId)); + if (!loop || candidate.nodeId === loop.batchNodeId) return false; + + const batchStep = steps[stepKeyId({ nodeId: loop.batchNodeId, iteration: candidate.iteration })]; + return batchStep !== undefined && isTerminalStep(batchStep); } /** Rule 2. A slot beyond the produced list reads undefined — dead, like null. */ -function isLiveEdge( - edge: GraphEdge, - iteration: number, - steps: Record, -): boolean { - const source = steps[stepKeyId({ nodeId: edge.from, iteration })]; - return source?.status === 'completed' && Boolean(source.filledOutputSlots[edge.outputIndex]); +function isLive(source: StepSummary, edge: GraphEdge): boolean { + return source.status === 'completed' && Boolean(source.filledOutputSlots[edge.outputIndex]); +} + +/** + * The steps `decideSuccessors` reads for this settlement: + * + * - the settled step, which for a batch node says which side to decide + * - each candidate successor, since an existing one means it was decided already + * - the steps each candidate's own incoming edges read + * - for a candidate inside a loop, that loop's batch step at the same iteration + * + * Derived from the same mapping the decision uses, so a caller cannot load a + * different set from the one that gets read. + */ +export function decisionKeys( + graph: WorkflowGraph, + loops: WorkflowLoop[], + settled: StepKey, + terminalIterations: Map, +): StepKey[] { + const keys = new Map([[stepKeyId(settled), settled]]); + const add = (key: StepKey) => keys.set(stepKeyId(key), key); + + const outgoingEdges = graph.edges.filter((edge) => edge.from === settled.nodeId); + + for (const edge of outgoingEdges) { + const target = targetKey(edge, classifyEdge(edge, loops), settled); + add(target); + + const incomingEdges = graph.edges.filter((inEdge) => inEdge.to === target.nodeId); + + for (const inEdge of incomingEdges) { + const source = sourceRow( + inEdge, + classifyEdge(inEdge, loops), + target, + terminalIterations.get(inEdge.from), + ); + if (source.kind === 'row') add(source.key); + } + + const loop = loops.find((l) => l.memberIds.has(target.nodeId)); + if (loop) add({ nodeId: loop.batchNodeId, iteration: target.iteration }); + } + + return [...keys.values()]; } diff --git a/packages/@n8n/engine/src/execution/step-ready-handler.ts b/packages/@n8n/engine/src/execution/step-ready-handler.ts index c33dabd80e8..c440c5ad9aa 100644 --- a/packages/@n8n/engine/src/execution/step-ready-handler.ts +++ b/packages/@n8n/engine/src/execution/step-ready-handler.ts @@ -1,9 +1,17 @@ import { UnexpectedError, UnimplementedError, type JsonValue } from '../common'; import type { ExternalDependencies, IStepExecutor } from '../dependencies'; -import type { GraphEdge, GraphNode } from '../graph'; +import { deriveLoops, type GraphEdge, type GraphNode, type WorkflowLoop } from '../graph'; import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; -import { stepKeyId, isSettledStatus, type StepKeyId, type StepSlots } from './execution.types'; +import { + stepKeyId, + isSettledStatus, + type StepKey, + type StepKeyId, + type StepSlots, +} from './execution.types'; +import { classifyEdge, sourceRow } from './iteration-mapping'; +import { exitSourcesInto, loadTerminalIterations } from './loop-ledger'; import type { StepError, StepRecord, StepStore } from './step-store'; import { validateStepContext } from './validate-step-context'; @@ -102,7 +110,7 @@ export class StepReadyHandler { return outputs; } - /** Inputs for `node`, each slot taken from its predecessor's output. */ + /** Inputs for `node`, each slot taken from the row its edge reads. */ private async gatherInputs(execution: ExecutionRecord, step: StepRecord): Promise { // These are all the edges that feed into the node this step runs. const incomingEdges = execution.graph.edges.filter((edge) => edge.to === step.nodeId); @@ -114,23 +122,31 @@ export class StepReadyHandler { ); } - validateIncomingEdges(incomingEdges, step); - - const predecessorNodeIds = [...new Set(incomingEdges.map((edge) => edge.from))]; - const predecessorSteps = await this.stepStore.loadStepsByKeys( + const loops = deriveLoops(execution.graph); + // Only an exit edge reads the row that ended a loop, so a step with no exit + // edge needs no such read at all. + const terminalIterations = await loadTerminalIterations( + this.stepStore, execution.id, - predecessorNodeIds.map((nodeId) => ({ nodeId, iteration: step.iteration })), + exitSourcesInto(execution.graph, loops, [step.nodeId]), ); + const reads = resolveInputReads(incomingEdges, loops, step, terminalIterations); + + // One key per distinct row: a predecessor wired to two input slots is read + // twice but loaded once. + const rows = await this.stepStore.loadStepsByKeys(execution.id, [ + ...new Map(reads.map(({ key }) => [stepKeyId(key), key])).values(), + ]); // 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 }, + { length: Math.max(...reads.map(({ edge }) => edge.inputIndex)) + 1 }, () => null, ); - for (const edge of incomingEdges) { - inputs[edge.inputIndex] = readEdgeValue(edge, step, predecessorSteps); + for (const { edge, key } of reads) { + inputs[edge.inputIndex] = readEdgeValue(edge, key, step, rows); } return inputs; @@ -155,11 +171,44 @@ export class StepReadyHandler { } } -// Validate that the incoming edges meet 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 { +/** + * Which row each incoming edge reads for this step, one per input slot. + * + * An edge that connects nothing at this iteration is dropped, which is what lets + * a batch node carry both an entry edge and a return edge on slot 0: they never + * apply at the same iteration, so per iteration the slot still has one source. + * Two edges that do both apply are the unsupported convergence case. + */ +export function resolveInputReads( + incomingEdges: GraphEdge[], + loops: WorkflowLoop[], + step: Pick, + terminalIterations: Map, +): Array<{ edge: GraphEdge; key: StepKey }> { + const reads = incomingEdges.flatMap((edge) => { + const source = sourceRow( + edge, + classifyEdge(edge, loops), + step, + terminalIterations.get(edge.from), + ); + if (source.kind === 'row') return [{ edge, key: source.key }]; + if (source.kind === 'none') return []; + // The planner queues a step only once every row it reads exists, so a loop + // that has not ended means the rows and the plan disagree. + throw new UnexpectedError( + `step ${step.id} reads node ${edge.from} across a loop that has not ended`, + ); + }); + + if (reads.length === 0) { + throw new UnexpectedError( + `step ${step.id} runs node ${step.nodeId}, which no edge reaches at iteration ${step.iteration}`, + ); + } + const filledSlots: Set = new Set(); - for (const edge of incomingEdges) { + for (const { edge } of reads) { if (filledSlots.has(edge.inputIndex)) { // TODO(CAT-3982): same-slot convergence gets a defined meaning. We // should have rejected this graph at validation time. @@ -169,6 +218,8 @@ function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): vo } filledSlots.add(edge.inputIndex); } + + return reads; } /** @@ -178,10 +229,11 @@ function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): vo */ function readEdgeValue( edge: GraphEdge, + source: StepKey, step: StepRecord, - predecessorSteps: Record, + rows: Record, ): JsonValue { - const row = predecessorSteps[stepKeyId({ nodeId: edge.from, iteration: step.iteration })]; + const row = rows[stepKeyId(source)]; if (!row || !isSettledStatus(row.status)) { // A step is planned only once every predecessor settled, so running on // a fabricated empty input would mask a planner/store inconsistency. diff --git a/packages/@n8n/engine/src/execution/step-settled-handler.ts b/packages/@n8n/engine/src/execution/step-settled-handler.ts index d493eeeafcb..7ad4216f4c8 100644 --- a/packages/@n8n/engine/src/execution/step-settled-handler.ts +++ b/packages/@n8n/engine/src/execution/step-settled-handler.ts @@ -1,22 +1,19 @@ import { UnexpectedError } from '../common'; -import { - findTriggerNode, - getDescendantNodeIds, - getPredecessorNodeIds, - getSuccessorNodeIds, -} from '../graph'; +import { deriveLoops, findTriggerNode, getDescendantNodeIds, getSuccessorNodeIds } from '../graph'; import type { OrchestrationMessage, StepMessage, StepSettledEvent, WorkQueue } from '../queue'; +import { countExpectedSettledSteps } from './completion'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; import { stepKeyId, type StepKey, type StepKeyId } from './execution.types'; -import { decideSuccessors } from './settlement'; -import type { StepRecord, StepStore, StepSummary } from './step-store'; +import { exitSourcesInto, loadTerminalIterations } from './loop-ledger'; +import { decideSuccessors, decisionKeys } from './settlement'; +import type { StepRecord, StepStore } from './step-store'; import { validateStepContext } from './validate-step-context'; /** * Handles the `step:settled` orchestration event: decides the fate of the * settled step's direct successors — queued when a live edge feeds them, * skipped when every input is settled dead (see the rules in `settlement.ts`) - * — and records the execution's outcome once every reachable node has + * — and records the execution's outcome once every step the execution owes has * settled. Skips are settlements too: each one is announced back onto the * orchestration queue, and handling it here decides the next hop, so a dead * region cascades through the event loop one settlement at a time. @@ -57,8 +54,7 @@ export class StepSettledHandler { return; } - const steps = await this.loadDecisionSteps(execution, step); - queued = await this.planSuccessors(execution, step, steps); + queued = await this.planSuccessors(execution, step); } // If we've queued steps, we know the execution isn't done yet, so we @@ -74,12 +70,27 @@ export class StepSettledHandler { } /** Plans the settled step's direct successors, returning how many were queued. */ - private async planSuccessors( - execution: ExecutionRecord, - step: StepRecord, - steps: Record, - ): Promise { - const { toQueue, toSkip } = decideSuccessors(execution.graph, step, steps); + private async planSuccessors(execution: ExecutionRecord, step: StepRecord): Promise { + const loops = deriveLoops(execution.graph); + // Only the candidates' own edges are resolved, so only the loops those + // edges leave need their latest row read. + const candidates = getSuccessorNodeIds(execution.graph, step.nodeId); + const terminalIterations = await loadTerminalIterations( + this.stepStore, + execution.id, + exitSourcesInto(execution.graph, loops, candidates), + ); + const steps = await this.stepStore.loadStepSummariesByKeys( + execution.id, + decisionKeys(execution.graph, loops, step, terminalIterations), + ); + const { toQueue, toSkip } = decideSuccessors( + execution.graph, + loops, + step, + steps, + terminalIterations, + ); if (toQueue.length === 0 && toSkip.length === 0) return 0; // One batch, so a settlement's consequence lands atomically and a fan-out @@ -96,28 +107,6 @@ export class StepSettledHandler { return await this.announceCreatedSteps(execution.id, created, new Set(toQueue.map(stepKeyId))); } - /** - * The rows a successor decision reads: the successors themselves (an - * existing row means already decided) and their predecessors (settledness - * and slot liveness), which include the settled node itself. - */ - private async loadDecisionSteps( - execution: ExecutionRecord, - step: StepRecord, - ): Promise> { - const successors = getSuccessorNodeIds(execution.graph, step.nodeId); - const nodeIds = [ - ...new Set([ - ...successors, - ...successors.flatMap((id) => getPredecessorNodeIds(execution.graph, id)), - ]), - ]; - // Forward edges stay within a pass, so every row read sits at the - // settled row's iteration. - const keys = nodeIds.map((nodeId) => ({ nodeId, iteration: step.iteration })); - return await this.stepStore.loadStepSummariesByKeys(execution.id, keys); - } - /** * Announces the created rows — `step:ready` for queued ones, `step:settled` * for skips, which settle at birth — and returns how many were queued. @@ -141,27 +130,43 @@ export class StepSettledHandler { } /** - * Records the execution's outcome once every reachable node has settled: - * `failed` if any step failed, `completed` otherwise. Settled rows are - * unique per node, only exist for reachable nodes, and never unsettle, so + * Records the execution's outcome once every step it owes has settled: `failed` + * if any step failed, `completed` otherwise. Steps are unique per + * `(node, iteration)`, only exist for reachable nodes, and never unsettle, so * the count comparison cannot pass early — in-flight events and unplanned - * successors both leave reachable nodes unsettled. + * successors both leave steps outstanding. */ private async finishExecutionIfDone(execution: ExecutionRecord): Promise { + const reachable = this.reachableNodeIds(execution); + const loops = deriveLoops(execution.graph); + const terminalIterations = await loadTerminalIterations( + this.stepStore, + execution.id, + // A loop the trigger cannot reach never receives rows, and the count below + // leaves it out, so its tip is not worth asking for. + loops + .filter((loop) => reachable.has(loop.batchNodeId)) + .map((loop) => loop.batchNodeId), + ); + const expected = countExpectedSettledSteps(loops, reachable, terminalIterations); + // A loop still running owes an unknown number of steps, so there is nothing + // to compare against yet. + if (expected === undefined) return; + const settled = await this.stepStore.countSettledSteps(execution.id); - if (settled < this.reachableNodeCount(execution)) return; + if (settled < expected) return; const failed = await this.stepStore.hasFailedSteps(execution.id); await this.executionStore.finishExecution(execution.id, failed ? 'failed' : 'completed'); } - private reachableNodeCount(execution: ExecutionRecord): number { + private reachableNodeIds(execution: ExecutionRecord): Set { const trigger = findTriggerNode(execution.graph); if (!trigger) { // The start boundary rejects triggerless graphs, so this execution // should never have been created. throw new UnexpectedError(`Execution ${execution.id} has no trigger node in its graph`); } - return 1 + getDescendantNodeIds(execution.graph, trigger.id).length; + return new Set([trigger.id, ...getDescendantNodeIds(execution.graph, trigger.id)]); } } diff --git a/packages/@n8n/engine/src/execution/step-store.ts b/packages/@n8n/engine/src/execution/step-store.ts index 769c5dc309b..42e7c5b1ff9 100644 --- a/packages/@n8n/engine/src/execution/step-store.ts +++ b/packages/@n8n/engine/src/execution/step-store.ts @@ -139,16 +139,26 @@ export interface StepStore { ): Promise>; /** - * The node's highest-iteration row, or `null` when it has none. For a - * batch node this is the loop's ledger tip. + * Planning view of each named node's highest-iteration row, keyed by node id, + * omitting the nodes with no row. For a batch node this is the row that says + * whether its loop has ended. + * + * One query for every node asked about, since a settlement can span several + * loops. The row ending a loop holds everything that loop accumulated, so this + * returns the same slim view as `loadStepSummariesByKeys`, not the whole row. */ - loadLatestStep(executionId: string, nodeId: string): Promise; + loadLatestStepSummaries( + executionId: string, + nodeIds: string[], + ): Promise>; /** * How many of the execution's steps have settled (completed, failed, - * skipped, or cancelled). Rows are unique per node and only exist for - * reachable nodes, so comparing this against the graph's reachable node - * count answers "has everything settled?" exactly. + * skipped, or cancelled). Rows are unique per `(node, iteration)`, only exist + * for reachable nodes, and never unsettle, so comparing this against the + * number of rows the execution owes answers "has everything settled?" exactly. + * A loop makes that number more than the node count, so the comparison runs + * against `expectedSettledRows` rather than against the graph. */ countSettledSteps(executionId: string): Promise;