mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat: Emit trigger payloads as step output slots (no-changelog) (#36416)
This commit is contained in:
parent
fdbf88b469
commit
0deba44cb8
|
|
@ -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(),
|
||||
}),
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -29,7 +29,9 @@ export class TypeOrmExecutionStore implements ExecutionStore {
|
|||
}
|
||||
|
||||
async loadExecution(id: string): Promise<ExecutionRecord> {
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ function record(graph: WorkflowGraph, overrides: Partial<ExecutionRecord> = {}):
|
|||
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: [] },
|
||||
]);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ function makeExecutionStore(overrides: Partial<ExecutionRecord> = {}): 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,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ function makeExecutionStore(
|
|||
status: 'running',
|
||||
mode: 'production',
|
||||
graph,
|
||||
triggerPayload: null,
|
||||
triggerOutputs: null,
|
||||
...overrides,
|
||||
};
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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 = [];
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -60,6 +60,7 @@ export type {
|
|||
StepSlots,
|
||||
StepStatus,
|
||||
StepStore,
|
||||
TriggerOutputs,
|
||||
} from './execution';
|
||||
|
||||
export {
|
||||
|
|
|
|||
|
|
@ -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')
|
||||
|
|
|
|||
|
|
@ -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<JsonValue> = z.lazy(() =>
|
||||
z.union([
|
||||
|
|
@ -17,8 +19,6 @@ const jsonValueSchema: z.ZodType<JsonValue> = 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(),
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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<typeof createDataSource>;
|
||||
|
||||
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<OrchestrationMessage>();
|
||||
|
|
@ -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([
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user