From 0deba44cb80d98bf0ed5367153cbc698d6de5980 Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Tue, 18 Aug 2026 12:29:18 +0000 Subject: [PATCH] feat: Emit trigger payloads as step output slots (no-changelog) (#36416) --- .../workflow-execution.integration.test.ts | 13 +++-- ...orkflow-step-execution.integration.test.ts | 2 +- .../entities/workflow-execution.entity.ts | 11 ++-- .../1778529600000-CreateWorkflowExecution.ts | 2 +- .../src/database/typeorm-execution-store.ts | 4 +- .../__tests__/execution-start-handler.test.ts | 45 ++++++++++++---- .../execution-start.integration.test.ts | 13 +++-- .../__tests__/start-execution.service.test.ts | 8 +-- .../step-execution.integration.test.ts | 43 +++++++-------- .../__tests__/step-ready-handler.test.ts | 4 +- .../__tests__/step-settled-handler.test.ts | 2 +- .../src/execution/execution-start-handler.ts | 12 ++--- .../engine/src/execution/execution-store.ts | 7 ++- .../engine/src/execution/execution.types.ts | 11 ++++ packages/@n8n/engine/src/execution/index.ts | 8 ++- .../src/execution/start-execution.service.ts | 8 +-- packages/@n8n/engine/src/graph/index.ts | 6 ++- .../src/graph/validate-executable-graph.ts | 2 +- packages/@n8n/engine/src/index.ts | 1 + .../workflow-executions.integration.test.ts | 54 +++++++++++++++++-- .../src/server/routes/workflow-executions.ts | 9 ++-- .../src/__tests__/acceptance-fixtures.ts | 6 +-- .../src/__tests__/io.test.ts | 8 +-- .../m1-acceptance.integration.test.ts | 44 ++++++++++++--- .../@n8n/node-engine-compatibility/src/io.ts | 3 -- 25 files changed, 229 insertions(+), 97 deletions(-) diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts index d22d5f8aed8..39983c8cf1b 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts @@ -30,18 +30,21 @@ describe('workflow_execution table (integration)', () => { status: 'running', mode: 'production', graph: { nodes: [], edges: [] }, - triggerPayload: { foo: 'bar' }, + triggerOutputs: [{ foo: 'bar' }], finishedAt: null, }); await repo.save(created); - const found = await repo.findOneByOrFail({ id: created.id }); + // NOTE: `findOne({ where })`, not `findOneByOrFail`: the latter's overload + // exceeds TypeScript's instantiation depth on the recursive `triggerOutputs` + // column type. + const found = await repo.findOneOrFail({ where: { id: created.id } }); expect(found.id).toBeTruthy(); expect(found.workflowId).toBe('wf-1'); expect(found.status).toBe('running'); expect(found.mode).toBe('production'); - expect(found.triggerPayload).toEqual({ foo: 'bar' }); + expect(found.triggerOutputs).toEqual([{ foo: 'bar' }]); expect(found.finishedAt).toBeNull(); expect(found.createdAt).toBeInstanceOf(Date); expect(found.updatedAt).toBeInstanceOf(Date); @@ -56,7 +59,7 @@ describe('workflow_execution table (integration)', () => { status: 'running', mode: 'production', graph: { nodes: [], edges: [] }, - triggerPayload: null, + triggerOutputs: null, finishedAt: null, }), ); @@ -66,7 +69,7 @@ describe('workflow_execution table (integration)', () => { status: 'completed', mode: 'production', graph: { nodes: [], edges: [] }, - triggerPayload: null, + triggerOutputs: null, finishedAt: new Date(), }), ); 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 dc26532b1a8..c32ba3db970 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 @@ -34,7 +34,7 @@ describe('workflow_step_execution table (integration)', () => { status: 'running', mode: 'production', graph: { nodes: [], edges: [] }, - triggerPayload: null, + triggerOutputs: null, finishedAt: null, }); await repo.save(execution); diff --git a/packages/@n8n/engine/src/database/entities/workflow-execution.entity.ts b/packages/@n8n/engine/src/database/entities/workflow-execution.entity.ts index 9ba841009e4..9debbb82512 100644 --- a/packages/@n8n/engine/src/database/entities/workflow-execution.entity.ts +++ b/packages/@n8n/engine/src/database/entities/workflow-execution.entity.ts @@ -8,8 +8,11 @@ import { UpdateDateColumn, } from '@n8n/typeorm'; -import type { JsonObject } from '../../common'; -import type { ExecutionMode, ExecutionStatus } from '../../execution/execution.types'; +import type { + ExecutionMode, + ExecutionStatus, + TriggerOutputs, +} from '../../execution/execution.types'; import type { WorkflowGraph } from '../../graph'; import { generateId } from '../generate-id'; @@ -32,8 +35,8 @@ export class WorkflowExecution { @Column('jsonb') graph!: WorkflowGraph; - @Column('jsonb', { name: 'trigger_payload', nullable: true }) - triggerPayload!: JsonObject | null; + @Column('jsonb', { name: 'trigger_outputs', nullable: true }) + triggerOutputs!: TriggerOutputs | null; @CreateDateColumn({ name: 'created_at', type: 'timestamptz', precision: 3 }) createdAt!: Date; diff --git a/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts b/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts index b6f61a3b204..5fd0f8cd56f 100644 --- a/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts +++ b/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts @@ -14,7 +14,7 @@ export class CreateWorkflowExecution1778529600000 implements MigrationInterface { name: 'status', type: 'varchar', length: '32' }, { name: 'mode', type: 'varchar', length: '32' }, { name: 'graph', type: 'jsonb' }, - { name: 'trigger_payload', type: 'jsonb', isNullable: true }, + { name: 'trigger_outputs', type: 'jsonb', isNullable: true }, { name: 'created_at', type: 'timestamptz', diff --git a/packages/@n8n/engine/src/database/typeorm-execution-store.ts b/packages/@n8n/engine/src/database/typeorm-execution-store.ts index 2bc9bdbc3c8..b585dcf13b2 100644 --- a/packages/@n8n/engine/src/database/typeorm-execution-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-execution-store.ts @@ -29,7 +29,9 @@ export class TypeOrmExecutionStore implements ExecutionStore { } async loadExecution(id: string): Promise { - const row = await this.repo.findOneBy({ id }); + // NOTE: `findOne({ where })`, not `findOneBy`: the latter's overload exceeds + // TypeScript's instantiation depth on the recursive `triggerOutputs` column type. + const row = await this.repo.findOne({ where: { id } }); if (!row) throw new ExecutionNotFoundError(id); return row; } 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 7af72c3e421..3e787482637 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 @@ -43,7 +43,7 @@ function record(graph: WorkflowGraph, overrides: Partial = {}): status: 'running', mode: 'production', graph, - triggerPayload: null, + triggerOutputs: null, ...overrides, }; } @@ -60,7 +60,7 @@ describe('ExecutionStartHandler', () => { const executionStore = makeExecutionStore({ loadExecution: vi .fn() - .mockResolvedValue(record(graph, { triggerPayload: { webhook: 'data' } })), + .mockResolvedValue(record(graph, { triggerOutputs: [[{ json: { webhook: 'data' } }]] })), }); const createSteps = vi.fn().mockResolvedValue([{ id: 'step-trigger', nodeId: 'trigger' }]); const stepStore = makeStepStore(createSteps); @@ -71,13 +71,13 @@ describe('ExecutionStartHandler', () => { expect(executionStore.transitionStatus).toHaveBeenCalledWith('exec-1', 'queued', 'running'); // only the trigger's row — planning is the step:settled handler's job. - // Its payload rides along as output slot 0, read downstream like any + // Its payload rides along as output slots, read downstream like any // other predecessor's outputs. expect(createSteps).toHaveBeenCalledExactlyOnceWith('exec-1', [ { nodeId: 'trigger', status: 'completed', - outputs: [{ webhook: 'data' }], + outputs: [[{ json: { webhook: 'data' } }]], }, ]); expect(queue.publish).toHaveBeenCalledExactlyOnceWith({ @@ -87,15 +87,17 @@ describe('ExecutionStartHandler', () => { }); }); - it('records an empty object in slot 0 when the trigger has no payload', async () => { - // The trigger fired — that is why the execution exists — so its output - // slot must count as data, or branching would treat it as never taken. + it('records multi-slot trigger payloads verbatim, including dead slots', async () => { const graph: WorkflowGraph = { nodes: [{ id: 'trigger', name: 'T', type: 'trigger' }], edges: [], }; const executionStore = makeExecutionStore({ - loadExecution: vi.fn().mockResolvedValue(record(graph, { triggerPayload: null })), + loadExecution: vi.fn().mockResolvedValue( + record(graph, { + triggerOutputs: [[{ json: { i1: true } }], null, [{ json: { i2: true } }]], + }), + ), }); const createSteps = vi.fn().mockResolvedValue([{ id: 'step-trigger', nodeId: 'trigger' }]); const stepStore = makeStepStore(createSteps); @@ -104,7 +106,32 @@ describe('ExecutionStartHandler', () => { await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' }); expect(createSteps).toHaveBeenCalledExactlyOnceWith('exec-1', [ - { nodeId: 'trigger', status: 'completed', outputs: [{}] }, + { + nodeId: 'trigger', + status: 'completed', + outputs: [[{ json: { i1: true } }], null, [{ json: { i2: true } }]], + }, + ]); + }); + + it('records no slots at all when the trigger has no payload', async () => { + // No payload means no data on any slot — every successor edge reads + // undefined and is treated as dead, same as any other step producing nothing. + const graph: WorkflowGraph = { + nodes: [{ id: 'trigger', name: 'T', type: 'trigger' }], + edges: [], + }; + const executionStore = makeExecutionStore({ + loadExecution: vi.fn().mockResolvedValue(record(graph, { triggerOutputs: null })), + }); + const createSteps = vi.fn().mockResolvedValue([{ id: 'step-trigger', nodeId: 'trigger' }]); + const stepStore = makeStepStore(createSteps); + const handler = new ExecutionStartHandler(executionStore, stepStore, makeOrchestrationQueue()); + + await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' }); + + expect(createSteps).toHaveBeenCalledExactlyOnceWith('exec-1', [ + { nodeId: 'trigger', status: 'completed', outputs: [] }, ]); }); diff --git a/packages/@n8n/engine/src/execution/__tests__/execution-start.integration.test.ts b/packages/@n8n/engine/src/execution/__tests__/execution-start.integration.test.ts index ab335f61a0e..2dfd82e094b 100644 --- a/packages/@n8n/engine/src/execution/__tests__/execution-start.integration.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/execution-start.integration.test.ts @@ -84,13 +84,15 @@ describe('execution start (integration)', () => { const { executionId } = await startExecution.start({ workflowId: 'wf-1', graph, - triggerPayload: null, + triggerOutputs: [[{ json: { hello: 'world' } }]], }); await ready; + // `findOne({ where })`, not `findOneByOrFail`: the latter's overload exceeds + // TypeScript's instantiation depth on the recursive `triggerOutputs` column type. const row = await dataSource .getRepository(WorkflowExecution) - .findOneByOrFail({ id: executionId }); + .findOneOrFail({ where: { id: executionId } }); expect(row.status).toBe('running'); const steps = await dataSource @@ -99,6 +101,7 @@ describe('execution start (integration)', () => { const triggerStep = steps.find((s) => s.nodeId === 'trigger'); const firstStep = steps.find((s) => s.nodeId === 'step-a'); expect(triggerStep?.status).toBe('completed'); + expect(triggerStep?.outputs).toEqual([[{ json: { hello: 'world' } }]]); expect(firstStep?.status).toBe('queued'); // step:ready references the durable step-record id, not the node id. @@ -119,7 +122,7 @@ describe('execution start (integration)', () => { status: 'queued', mode: 'production', graph, - triggerPayload: null, + triggerOutputs: null, }); // Delivered twice, both awaited — the CAS is what makes the second a no-op. @@ -127,9 +130,11 @@ describe('execution start (integration)', () => { await handler.handle(event); await handler.handle(event); + // `findOne({ where })`, not `findOneByOrFail`: the latter's overload exceeds + // TypeScript's instantiation depth on the recursive `triggerOutputs` column type. const row = await dataSource .getRepository(WorkflowExecution) - .findOneByOrFail({ id: executionId }); + .findOneOrFail({ where: { id: executionId } }); expect(row.status).toBe('running'); const steps = await dataSource diff --git a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts index 664e898750a..48dd8fd3b4f 100644 --- a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts @@ -37,7 +37,7 @@ describe('StartExecutionService', () => { const result = await service.start({ workflowId: 'wf-1', graph: sampleGraph, - triggerPayload: { hello: 'world' }, + triggerOutputs: [[{ json: { hello: 'world' } }]], }); expect(result.executionId).toBe('exec-id-1'); @@ -47,7 +47,7 @@ describe('StartExecutionService', () => { status: 'queued', mode: 'production', graph: sampleGraph, - triggerPayload: { hello: 'world' }, + triggerOutputs: [[{ json: { hello: 'world' } }]], }); expect(queue.publish).toHaveBeenCalledWith({ type: 'execution:enqueued', @@ -55,7 +55,7 @@ describe('StartExecutionService', () => { }); }); - it('defaults mode to production and triggerPayload to null', async () => { + it('defaults mode to production and triggerOutputs to null', async () => { const admittance: AdmittanceService = { evaluate: vi.fn().mockResolvedValue({ accept: true }), }; @@ -66,7 +66,7 @@ describe('StartExecutionService', () => { await service.start({ workflowId: 'wf-1', graph: sampleGraph }); expect(store.createExecution).toHaveBeenCalledWith( - expect.objectContaining({ mode: 'production', triggerPayload: null }), + expect.objectContaining({ mode: 'production', triggerOutputs: null }), ); }); 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 67afc9a26cd..cf81b77ee46 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 @@ -4,7 +4,6 @@ import postgresVersions from 'n8n-containers/postgres-versions.json'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { AllowAllAdmittance } from '../../admittance'; -import type { JsonObject } from '../../common'; import { createDataSource, TypeOrmExecutionStore, @@ -16,6 +15,7 @@ import type { IStepExecutor, StepExecutionRequest } from '../../dependencies'; import type { WorkflowGraph } from '../../graph'; import { InMemoryWorkQueue, type OrchestrationMessage, type StepMessage } from '../../queue'; import { ExecutionStartHandler } from '../execution-start-handler'; +import type { TriggerOutputs } from '../execution.types'; import { OrchestrationWorker } from '../orchestration-worker'; import { StartExecutionService } from '../start-execution.service'; import { StepReadyHandler } from '../step-ready-handler'; @@ -60,7 +60,7 @@ describe('step execution (integration)', () => { */ async function runWorkflow( executor: IStepExecutor, - triggerPayload: JsonObject, + triggerOutputs: TriggerOutputs, { workflowId = 'wf-1', graph: workflowGraph = graph } = {}, ) { const { executionStore, stepStore } = stores(); @@ -94,7 +94,7 @@ describe('step execution (integration)', () => { new AllowAllAdmittance(), executionStore, orchestrationQueue, - ).start({ workflowId, graph: workflowGraph, triggerPayload }); + ).start({ workflowId, graph: workflowGraph, triggerOutputs }); await finished; await stepWorker.stop(); @@ -119,9 +119,9 @@ describe('step execution (integration)', () => { }, }; - const { executionId, execution, steps } = await runWorkflow(executor, { - body: { name: 'ada' }, - }); + const { executionId, execution, steps } = await runWorkflow(executor, [ + { body: { name: 'ada' } }, + ]); const step = steps.find(({ nodeId }) => nodeId === 'node-a'); expect(execution.status).toBe('completed'); @@ -150,7 +150,7 @@ describe('step execution (integration)', () => { }, }; - const { execution, steps } = await runWorkflow(executor, {}); + const { execution, steps } = await runWorkflow(executor, [{}]); const step = steps.find(({ nodeId }) => nodeId === 'node-a'); // the failure is terminal for the execution too @@ -187,11 +187,10 @@ describe('step execution (integration)', () => { }, }; - const { execution } = await runWorkflow( - executor, - { body: { name: 'ada' } }, - { workflowId: 'wf-chain', graph: chainGraph }, - ); + const { execution } = await runWorkflow(executor, [{ body: { name: 'ada' } }], { + workflowId: 'wf-chain', + graph: chainGraph, + }); // both nodes ran, in order, each on what came before it: node-a on the // trigger's payload slot, node-b on node-a's output slot 0 @@ -227,11 +226,10 @@ describe('step execution (integration)', () => { }, }; - const { execution } = await runWorkflow( - executor, - { body: { name: 'ada' } }, - { workflowId: 'wf-diamond', graph: diamondGraph }, - ); + 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'); @@ -277,11 +275,10 @@ describe('step execution (integration)', () => { }, }; - const { execution, steps } = await runWorkflow( - executor, - {}, - { workflowId: 'wf-branch', graph: branchingGraph }, - ); + const { execution, steps } = await runWorkflow(executor, [{}], { + workflowId: 'wf-branch', + graph: branchingGraph, + }); // the dead branch never ran a node expect(requests.map(({ node }) => node.id).sort()).toEqual(['node-a', 'node-if', 'node-m']); @@ -319,7 +316,7 @@ describe('step execution (integration)', () => { status: 'running', mode: 'production', graph, - triggerPayload: null, + triggerOutputs: null, }); const created = await stepStore.createSteps(executionId, [ // completed steps always carry outputs, as the start handler writes them 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 9fe7eabf7b1..dc8abd6391d 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 @@ -33,7 +33,7 @@ function makeExecutionStore(overrides: Partial = {}): Execution status: 'running', mode: 'production', graph, - triggerPayload: null, + triggerOutputs: null, ...overrides, }; return { @@ -95,7 +95,7 @@ describe('StepReadyHandler', () => { const executor = makeExecutor({ outputs: [[{ json: { ok: true } }]] }); // a stale payload on the execution record must not be consulted: the // trigger's step row is the one source of its output - const executionStore = makeExecutionStore({ triggerPayload: { body: { stale: true } } }); + const executionStore = makeExecutionStore({ triggerOutputs: [{ body: { stale: true } }] }); const handler = new StepReadyHandler(executionStore, stepStore, queue, { v1StepExecutor: executor, }); 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 8947beb65d0..cd32a381997 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 @@ -53,7 +53,7 @@ function makeExecutionStore( status: 'running', mode: 'production', graph, - triggerPayload: null, + triggerOutputs: null, ...overrides, }; return { diff --git a/packages/@n8n/engine/src/execution/execution-start-handler.ts b/packages/@n8n/engine/src/execution/execution-start-handler.ts index 8ae2a93a94e..07e23086eff 100644 --- a/packages/@n8n/engine/src/execution/execution-start-handler.ts +++ b/packages/@n8n/engine/src/execution/execution-start-handler.ts @@ -2,6 +2,7 @@ import { UnexpectedError } from '../common'; import { findTriggerNode } from '../graph'; import type { ExecutionEnqueuedEvent, OrchestrationMessage, WorkQueue } from '../queue'; import type { ExecutionStore } from './execution-store'; +import { DEFAULT_TRIGGER_OUTPUTS } from './execution.types'; import type { StepStore } from './step-store'; /** @@ -37,17 +38,16 @@ export class ExecutionStartHandler { throw new UnexpectedError(`Execution ${event.executionId} has no trigger node in its graph`); } - // The trigger's output was captured at execution start; record it as - // already completed so successors can treat it as a satisfied predecessor. + // The trigger's outputs were captured at execution start; record them as + // already completed so successors read them like any predecessor's slots. + // No payload means no slots at all: every successor edge reads undefined + // and is treated as dead, same as any other step that produced nothing. // The claim above makes this the only writer, so the row cannot exist yet. - // NOTE: trigger payloads are not really supported yet — the raw payload is - // not item-shaped, so the v1 shim coerces it to zero items. Proper trigger - // handling will decide its shape. const [triggerStep] = await this.stepStore.createSteps(event.executionId, [ { nodeId: trigger.id, status: 'completed', - outputs: [execution.triggerPayload ?? {}], + outputs: execution.triggerOutputs ?? DEFAULT_TRIGGER_OUTPUTS, }, ]); if (!triggerStep) { diff --git a/packages/@n8n/engine/src/execution/execution-store.ts b/packages/@n8n/engine/src/execution/execution-store.ts index b310c4cd637..96243e57719 100644 --- a/packages/@n8n/engine/src/execution/execution-store.ts +++ b/packages/@n8n/engine/src/execution/execution-store.ts @@ -1,6 +1,5 @@ -import type { JsonObject } from '../common'; import type { WorkflowGraph } from '../graph'; -import type { ExecutionMode, ExecutionStatus } from './execution.types'; +import type { ExecutionMode, ExecutionStatus, TriggerOutputs } from './execution.types'; /** A new execution to persist. `id` and timestamps are assigned by the store. */ export interface NewExecutionRecord { @@ -8,7 +7,7 @@ export interface NewExecutionRecord { status: ExecutionStatus; mode: ExecutionMode; graph: WorkflowGraph; - triggerPayload: JsonObject | null; + triggerOutputs: TriggerOutputs | null; } /** A full execution record. */ @@ -18,7 +17,7 @@ export interface ExecutionRecord { status: ExecutionStatus; mode: ExecutionMode; graph: WorkflowGraph; - triggerPayload: JsonObject | null; + triggerOutputs: TriggerOutputs | null; } /** Thrown by `loadExecution` when no execution exists for the given id. */ diff --git a/packages/@n8n/engine/src/execution/execution.types.ts b/packages/@n8n/engine/src/execution/execution.types.ts index 8f85b886564..ead3f4085de 100644 --- a/packages/@n8n/engine/src/execution/execution.types.ts +++ b/packages/@n8n/engine/src/execution/execution.types.ts @@ -40,3 +40,14 @@ export function isSettledStatus(status: StepStatus): boolean { * the same thing — a step that ran and produced zero items is still live. */ export type StepSlots = JsonValue[]; + +/** + * The trigger step's outputs, supplied by whoever starts the execution: one + * entry per output slot, `null` for a slot the trigger didn't fire. Same shape + * and same opacity as any other step's `StepSlots` — a v1 host puts JSON-shaped + * `INodeExecutionData[]` in each slot. + */ +export type TriggerOutputs = StepSlots; + +/** Slots recorded for a trigger that fired without a payload: no slots at all. */ +export const DEFAULT_TRIGGER_OUTPUTS: TriggerOutputs = []; diff --git a/packages/@n8n/engine/src/execution/index.ts b/packages/@n8n/engine/src/execution/index.ts index 13d77b8ac2a..023a4d138f2 100644 --- a/packages/@n8n/engine/src/execution/index.ts +++ b/packages/@n8n/engine/src/execution/index.ts @@ -3,7 +3,13 @@ export type { StartExecutionRequest, StartExecutionResult, } from './start-execution.service'; -export type { ExecutionMode, ExecutionStatus, StepSlots, StepStatus } from './execution.types'; +export type { + ExecutionMode, + ExecutionStatus, + StepSlots, + StepStatus, + TriggerOutputs, +} from './execution.types'; export { ExecutionNotFoundError } from './execution-store'; export type { ExecutionRecord, ExecutionStore, NewExecutionRecord } from './execution-store'; export { StepNotFoundError } from './step-store'; diff --git a/packages/@n8n/engine/src/execution/start-execution.service.ts b/packages/@n8n/engine/src/execution/start-execution.service.ts index c25e9143d4d..fcf0c7b7b0a 100644 --- a/packages/@n8n/engine/src/execution/start-execution.service.ts +++ b/packages/@n8n/engine/src/execution/start-execution.service.ts @@ -1,14 +1,14 @@ import { AdmittanceRejectedError, type AdmittanceService } from '../admittance'; -import type { JsonObject } from '../common'; import { validateExecutableGraph, type WorkflowGraph } from '../graph'; import type { OrchestrationMessage, WorkQueue } from '../queue'; import type { ExecutionStore } from './execution-store'; -import type { ExecutionMode } from './execution.types'; +import type { ExecutionMode, TriggerOutputs } from './execution.types'; export interface StartExecutionRequest { workflowId: string; graph: WorkflowGraph; - triggerPayload?: JsonObject | null; + /** Trigger step's output slots, one entry per output. */ + triggerOutputs?: TriggerOutputs | null; mode?: ExecutionMode; } @@ -40,7 +40,7 @@ export class StartExecutionService { status: 'queued', mode: request.mode ?? 'production', graph: request.graph, - triggerPayload: request.triggerPayload ?? null, + triggerOutputs: request.triggerOutputs ?? null, }); // TODO(CAT-2938): the persist above and this publish aren't atomic — a diff --git a/packages/@n8n/engine/src/graph/index.ts b/packages/@n8n/engine/src/graph/index.ts index d78f8d0dc05..e0e0f82296f 100644 --- a/packages/@n8n/engine/src/graph/index.ts +++ b/packages/@n8n/engine/src/graph/index.ts @@ -11,4 +11,8 @@ export { getPredecessorNodeIds, getSuccessorNodeIds, } from './workflow-graph-queries'; -export { GraphValidationError, validateExecutableGraph } from './validate-executable-graph'; +export { + GraphValidationError, + MAX_SLOT_INDEX, + validateExecutableGraph, +} from './validate-executable-graph'; diff --git a/packages/@n8n/engine/src/graph/validate-executable-graph.ts b/packages/@n8n/engine/src/graph/validate-executable-graph.ts index 3bf3b1a7bb9..48b4017a428 100644 --- a/packages/@n8n/engine/src/graph/validate-executable-graph.ts +++ b/packages/@n8n/engine/src/graph/validate-executable-graph.ts @@ -2,7 +2,7 @@ import { UnimplementedError } from '../common'; import type { WorkflowGraph } from './workflow-graph'; import { getDescendantNodeIds } from './workflow-graph-queries'; -const MAX_SLOT_INDEX = 100; +export const MAX_SLOT_INDEX = 100; /** Thrown when a graph fails a structural rule and can never execute. */ export class GraphValidationError extends Error { diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index 0025bc5c4db..a91c5fb1d5e 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/packages/@n8n/engine/src/index.ts @@ -60,6 +60,7 @@ export type { StepSlots, StepStatus, StepStore, + TriggerOutputs, } from './execution'; export { diff --git a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts b/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts index 2f216d7761a..ad94638b176 100644 --- a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts +++ b/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts @@ -53,7 +53,7 @@ describe('POST /api/workflow-executions (integration)', () => { .send({ workflowId: 'wf-1', graph: sampleGraph, - triggerPayload: { hello: 'world' }, + triggerOutputs: [[{ json: { hello: 'world' } }]], }); expect(response.status).toBe(201); @@ -61,12 +61,14 @@ describe('POST /api/workflow-executions (integration)', () => { expect(executionId).toBeTruthy(); const repo = dataSource.getRepository(WorkflowExecution); - const row = await repo.findOneByOrFail({ id: executionId }); + // `findOne({ where })`, not `findOneByOrFail`: the latter's overload exceeds + // TypeScript's instantiation depth on the recursive `triggerOutputs` column type. + const row = await repo.findOneOrFail({ where: { id: executionId } }); expect(row.workflowId).toBe('wf-1'); expect(row.status).toBe('queued'); expect(row.mode).toBe('production'); expect(row.graph).toEqual(sampleGraph); - expect(row.triggerPayload).toEqual({ hello: 'world' }); + expect(row.triggerOutputs).toEqual([[{ json: { hello: 'world' } }]]); expect(workQueue.publish).toHaveBeenCalledWith({ type: 'execution:enqueued', executionId }); }); @@ -80,6 +82,52 @@ describe('POST /api/workflow-executions (integration)', () => { expect((response.body as { error: string }).error).toBe('invalid_request'); }); + it('rejects a bare-object triggerOutputs with 400 (the legacy, dropped shape)', async () => { + const response = await request(url) + .post('/api/workflow-executions') + .send({ + workflowId: 'wf-1', + graph: sampleGraph, + triggerOutputs: { hello: 'world' }, + }); + + expect(response.status).toBe(400); + expect((response.body as { error: string }).error).toBe('invalid_request'); + }); + + it.each([['str'], [42]])('rejects triggerOutputs %p with 400', async (triggerOutputs) => { + const response = await request(url) + .post('/api/workflow-executions') + .send({ workflowId: 'wf-1', graph: sampleGraph, triggerOutputs }); + + expect(response.status).toBe(400); + expect((response.body as { error: string }).error).toBe('invalid_request'); + }); + + it('rejects an empty-array triggerOutputs with 400 (send null or omit for "no payload")', async () => { + const response = await request(url).post('/api/workflow-executions').send({ + workflowId: 'wf-1', + graph: sampleGraph, + triggerOutputs: [], + }); + + expect(response.status).toBe(400); + expect((response.body as { error: string }).error).toBe('invalid_request'); + }); + + it('rejects a triggerOutputs with more slots than the cap with 400', async () => { + const response = await request(url) + .post('/api/workflow-executions') + .send({ + workflowId: 'wf-1', + graph: sampleGraph, + triggerOutputs: Array.from({ length: 102 }, () => []), + }); + + expect(response.status).toBe(400); + expect((response.body as { error: string }).error).toBe('invalid_request'); + }); + it('rejects a graph without a trigger with 400, creating nothing', async () => { const response = await request(url) .post('/api/workflow-executions') diff --git a/packages/@n8n/engine/src/server/routes/workflow-executions.ts b/packages/@n8n/engine/src/server/routes/workflow-executions.ts index c384c9dbff4..df0da705998 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.ts @@ -4,7 +4,9 @@ import { z } from 'zod'; import { AdmittanceRejectedError } from '../../admittance'; import { UnimplementedError, type JsonValue } from '../../common'; import type { StartExecutionService } from '../../execution/start-execution.service'; -import { GraphValidationError } from '../../graph'; +import { GraphValidationError, MAX_SLOT_INDEX } from '../../graph'; + +const MAX_TRIGGER_SLOTS = MAX_SLOT_INDEX + 1; const jsonValueSchema: z.ZodType = z.lazy(() => z.union([ @@ -17,8 +19,6 @@ const jsonValueSchema: z.ZodType = z.lazy(() => ]), ); -const jsonObjectSchema = z.record(jsonValueSchema); - const StepTypeSchema = z.enum(['trigger', 'v1-node', 'wait', 'subworkflow', 'batch']); const GraphNodeSchema = z.object({ @@ -46,7 +46,8 @@ const WorkflowGraphSchema = z.object({ const StartExecutionBody = z.object({ workflowId: z.string().min(1), graph: WorkflowGraphSchema, - triggerPayload: jsonObjectSchema.nullable().optional(), + /** Trigger output slots. Empty means "no payload" — send `null` or omit instead. */ + triggerOutputs: z.array(jsonValueSchema).min(1).max(MAX_TRIGGER_SLOTS).nullable().optional(), mode: z.enum(['production', 'manual']).optional(), }); diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts index 64328775f7f..c1afe81cc9d 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts @@ -1,8 +1,8 @@ import type { createDataSource, - JsonObject, OrchestrationMessage, StepMessage, + TriggerOutputs, WorkflowGraph, } from '@n8n/engine'; import { @@ -95,7 +95,7 @@ export function setWorkflow(assignments: Assignment[]) { type EngineDataSource = ReturnType; export function makeRunWorkflow(getDataSource: () => EngineDataSource) { - return async function runWorkflow(graph: WorkflowGraph, triggerPayload: JsonObject | null) { + return async function runWorkflow(graph: WorkflowGraph, triggerOutputs: TriggerOutputs | null) { const dataSource = getDataSource(); const { executionStore, stepStore } = createStores(dataSource); const orchestrationQueue = new InMemoryWorkQueue(); @@ -134,7 +134,7 @@ export function makeRunWorkflow(getDataSource: () => EngineDataSource) { new AllowAllAdmittance(), executionStore, orchestrationQueue, - ).start({ workflowId: 'wf-m1', graph, triggerPayload }); + ).start({ workflowId: 'wf-m1', graph, triggerOutputs }); try { await Promise.race([ diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/io.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/io.test.ts index d162620ae6b..32f3424b125 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/io.test.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/io.test.ts @@ -36,13 +36,13 @@ describe('fromStepInputs', () => { expect(fromStepInputs([null])).toEqual([[]]); }); - it('wraps a bare object slot as a single item (trigger payload shape)', () => { - expect(fromStepInputs([{ name: 'ada' }])).toEqual([[{ json: { name: 'ada' } }]]); - }); - it('yields an empty item list for a slot not carrying items', () => { expect(fromStepInputs(['nope'])).toEqual([[]]); }); + + it('yields an empty item list for a bare object slot (no longer a valid trigger shape)', () => { + expect(fromStepInputs([{ name: 'ada' }])).toEqual([[]]); + }); }); describe('toStepOutputs', () => { diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/m1-acceptance.integration.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/m1-acceptance.integration.test.ts index 229e5d09296..71b83b13a64 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/m1-acceptance.integration.test.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/m1-acceptance.integration.test.ts @@ -32,7 +32,7 @@ describe('M1 acceptance (integration)', () => { it('runs a single v1 node (Set) workflow to completion with correct rows', async () => { const graph = setWorkflow([{ name: 'greeting', value: 'hello', type: 'string' }]); - const { execution, steps, byNode } = await runWorkflow(graph, { name: 'ada' }); + const { execution, steps, byNode } = await runWorkflow(graph, [[{ json: { name: 'ada' } }]]); expect(steps).toHaveLength(2); expect(byNode('trigger')?.status).toBe('completed'); @@ -65,7 +65,7 @@ describe('M1 acceptance (integration)', () => { ), ); - const { execution, steps, byNode } = await runWorkflow(graph, { seed: 's' }); + const { execution, steps, byNode } = await runWorkflow(graph, [[{ json: { seed: 's' } }]]); expect(steps).toHaveLength(4); for (const nodeId of ['trigger', 'node-a', 'node-b', 'node-c']) { @@ -97,9 +97,9 @@ describe('M1 acceptance (integration)', () => { ), ); - const { execution, byNode } = await runWorkflow(graph, { - orders: [{ n: 1 }, { n: 2 }, { n: 3 }], - }); + const { execution, byNode } = await runWorkflow(graph, [ + [{ json: { orders: [{ n: 1 }, { n: 2 }, { n: 3 }] } }], + ]); expect(byNode('mark')?.outputs).toEqual([ [ @@ -120,18 +120,46 @@ describe('M1 acceptance (integration)', () => { type: 'string', }, ]); - const { set } = { set: (await runWorkflow(graph, { name: 'ada' })).byNode('set-node') }; + const { set } = { + set: (await runWorkflow(graph, [[{ json: { name: 'ada' } }]])).byNode('set-node'), + }; expect(set?.outputs).toEqual([ [expect.objectContaining({ json: { byInput: 'ada!', byReference: 'ada?' } })], ]); }); + it('emits one item per trigger payload item, with by-reference expressions still resolving', async () => { + const graph = setWorkflow([ + { name: 'byInput', value: '={{ $json.n }}', type: 'number' }, + { + name: 'byReference', + value: "={{ $('When clicking Execute').first().json.x }}", + type: 'string', + }, + ]); + const { byNode } = await runWorkflow(graph, [ + [ + { json: { x: 'from-trigger', n: 1 } }, + { json: { x: 'from-trigger', n: 2 } }, + { json: { x: 'from-trigger', n: 3 } }, + ], + ]); + + expect(byNode('set-node')?.outputs).toEqual([ + [ + expect.objectContaining({ json: { byInput: 1, byReference: 'from-trigger' } }), + expect.objectContaining({ json: { byInput: 2, byReference: 'from-trigger' } }), + expect.objectContaining({ json: { byInput: 3, byReference: 'from-trigger' } }), + ], + ]); + }); + it('fails the execution when a node errors, leaving no orphan step rows', async () => { const graph = setWorkflow([ { name: 'boom', value: "={{ $('Ghost').first().json.x }}", type: 'string' }, ]); - const { execution, steps, byNode } = await runWorkflow(graph, { name: 'ada' }); + const { execution, steps, byNode } = await runWorkflow(graph, [[{ json: { name: 'ada' } }]]); expect(steps).toHaveLength(2); expect(byNode('set-node')?.status).toBe('failed'); @@ -161,7 +189,7 @@ describe('M1 acceptance (integration)', () => { ), ); - const { execution, steps, byNode } = await runWorkflow(graph, {}); + const { execution, steps, byNode } = await runWorkflow(graph, [[]]); expect(byNode('node-a')?.status).toBe('failed'); expect(byNode('node-b')).toBeUndefined(); diff --git a/packages/@n8n/node-engine-compatibility/src/io.ts b/packages/@n8n/node-engine-compatibility/src/io.ts index 9de2d688516..4cf1d1834ab 100644 --- a/packages/@n8n/node-engine-compatibility/src/io.ts +++ b/packages/@n8n/node-engine-compatibility/src/io.ts @@ -5,9 +5,6 @@ import { isRecord } from './guards'; export function fromStepInputs(value: StepSlots): INodeExecutionData[][] { return value.map((items) => { - // a bare object in a slot is a single item: the shape of a trigger - // payload, until proper trigger handling decides its shape - if (isRecord(items)) return [{ json: items as IDataObject }]; if (!Array.isArray(items)) return []; return items.map((item): INodeExecutionData => { if (isRecord(item) && isRecord(item.json)) return item as unknown as INodeExecutionData;