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