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 400e8b77398..b8821908c77 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts @@ -1,5 +1,6 @@ import type { createDataSource, + ExecutionMode, StartExecutionResult, TriggerOutputs, WorkflowGraph, @@ -93,7 +94,11 @@ export function setWorkflow(assignments: Assignment[]) { type EngineDataSource = ReturnType; export function makeRunWorkflow(getDataSource: () => EngineDataSource) { - return async function runWorkflow(graph: WorkflowGraph, triggerOutputs: TriggerOutputs | null) { + return async function runWorkflow( + graph: WorkflowGraph, + triggerOutputs: TriggerOutputs | null, + mode?: ExecutionMode, + ) { const dataSource = getDataSource(); let done!: () => void; @@ -127,7 +132,7 @@ export function makeRunWorkflow(getDataSource: () => EngineDataSource) { const response = await request(runtime.app) .post('/api/workflow-executions') .set('Authorization', `Bearer ${mintIdentityToken(authSecret, caller)}`) - .send({ workflowId: 'wf-m1', graph, triggerOutputs }) + .send({ workflowId: 'wf-m1', graph, triggerOutputs, mode }) .expect(201); const { executionId } = response.body as StartExecutionResult; 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 71b83b13a64..40b8742990c 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 @@ -76,6 +76,35 @@ describe('M1 acceptance (integration)', () => { expect(execution.status).toBe('completed'); }); + it('runs a manual execution started with the default trigger payload', async () => { + const graph = converter.convert( + v1Workflow( + [ + TRIGGER, + setNode('node-a', 'A', [{ name: 'a', value: 'from-a', type: 'string' }]), + setNode('node-b', 'B', [{ name: 'b', value: '={{ $json.a }}-b', type: 'string' }]), + ], + { + 'When clicking Execute': mainTo('A'), + A: mainTo('B'), + }, + ), + ); + + // what the control plane sends for a manual run with no trigger data + const { execution, steps, byNode } = await runWorkflow(graph, [[{ json: {} }]], 'manual'); + + expect(steps).toHaveLength(3); + for (const nodeId of ['trigger', 'node-a', 'node-b']) { + expect(byNode(nodeId)?.status).toBe('completed'); + } + expect(byNode('node-b')?.outputs).toEqual([ + [expect.objectContaining({ json: { b: 'from-a-b' } })], + ]); + expect(execution.mode).toBe('manual'); + expect(execution.status).toBe('completed'); + }); + it('processes an n-item input into an n-item output, serially and in order', async () => { const graph = converter.convert( v1Workflow( diff --git a/packages/@n8n/node-engine-compatibility/src/index.ts b/packages/@n8n/node-engine-compatibility/src/index.ts index 7864f47a5b0..b4e5e7cf30a 100644 --- a/packages/@n8n/node-engine-compatibility/src/index.ts +++ b/packages/@n8n/node-engine-compatibility/src/index.ts @@ -1,5 +1,6 @@ export { V1WorkflowConverter } from './v1-workflow-converter'; export { V1StepExecutor } from './v1-step-executor'; export { createEngineStepDataLoader } from './engine-step-data-loader'; +export { toStepOutputs } from './io'; export { UnsupportedTriggerError, UnsupportedWorkflowError } from './errors'; export type { StepData, StepDataLoader, V1StepExecutorDeps } from './types'; diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 71fe9b737aa..4cca0cbb3fd 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/packages/cli/src/__tests__/workflow-runner.test.ts @@ -38,6 +38,7 @@ import { ExecutionNotFoundError } from '@/errors/execution-not-found-error'; import * as ExecutionLifecycleHooks from '@/execution-lifecycle/execution-lifecycle-hooks'; import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks'; import { ManualExecutionService } from '@/manual-execution.service'; +import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; import { OwnershipService } from '@/services/ownership.service'; import { Telemetry } from '@/telemetry'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; @@ -270,6 +271,57 @@ describe('run', () => { expect(addSpy).toHaveBeenCalledWith(data, existingExecution); }); + describe('engine 2.0 dispatch', () => { + it('hands the run to the dispatcher without registering a control-plane execution', async () => { + // ARRANGE + const dispatcher = Container.get(EngineV2Dispatcher); + vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(true); + const startSpy = vi.spyOn(dispatcher, 'start').mockResolvedValueOnce('dp-uuid'); + const addSpy = vi.spyOn(Container.get(ActiveExecutions), 'add'); + + const data = mock(); + + // ACT + const executionId = await runner.run(data); + + // ASSERT + expect(executionId).toBe('dp-uuid'); + expect(startSpy).toHaveBeenCalledWith(data); + expect(addSpy).not.toHaveBeenCalled(); + }); + + it('leaves the v1 path alone when the run does not route to engine 2.0', async () => { + // ARRANGE + const dispatcher = Container.get(EngineV2Dispatcher); + vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(false); + const startSpy = vi.spyOn(dispatcher, 'start'); + const activeExecutions = Container.get(ActiveExecutions); + const addSpy = vi.spyOn(activeExecutions, 'add').mockResolvedValue('1'); + vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValueOnce(); + vi.spyOn(Container.get(CredentialsPermissionChecker), 'check').mockResolvedValueOnce(); + vi.spyOn(WorkflowExecute.prototype, 'run').mockReturnValueOnce( + new PCancelable(() => mock()), + ); + + const data = mock({ + triggerToStartFrom: undefined, + workflowData: { nodes: [], staticData: {} }, + executionData: undefined, + startNodes: undefined, + destinationNode: undefined, + runData: undefined, + }); + + // ACT + const executionId = await runner.run(data); + + // ASSERT + expect(executionId).toBe('1'); + expect(startSpy).not.toHaveBeenCalled(); + expect(addSpy).toHaveBeenCalledWith(data, undefined); + }); + }); + it('run partial execution with additional data', async () => { // ARRANGE const activeExecutions = Container.get(ActiveExecutions); diff --git a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts index eeb89fb7a1d..dd841af7599 100644 --- a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts +++ b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts @@ -17,6 +17,14 @@ describe('EngineDataPlaneProxyService', () => { proxy = new EngineDataPlaneProxyService(); }); + it('is unavailable until a provider registers', () => { + expect(proxy.isAvailable()).toBe(false); + + proxy.registerProvider(mock()); + + expect(proxy.isAvailable()).toBe(true); + }); + it('explains how to enable the engine when no provider is registered', async () => { await expect(proxy.startExecution(request)).rejects.toThrow(UserError); await expect(proxy.startExecution(request)).rejects.toThrow('N8N_ENABLED_MODULES'); diff --git a/packages/cli/src/services/__tests__/engine-v2-dispatcher.service.test.ts b/packages/cli/src/services/__tests__/engine-v2-dispatcher.service.test.ts new file mode 100644 index 00000000000..3edc3dd2e87 --- /dev/null +++ b/packages/cli/src/services/__tests__/engine-v2-dispatcher.service.test.ts @@ -0,0 +1,321 @@ +import { UnsupportedTriggerError } from '@n8n/node-engine-compatibility'; +import type { + INode, + IPinData, + IRunData, + ITaskData, + ITaskDataConnections, + IWorkflowBase, + IWorkflowExecutionDataProcess, + StartNodeData, + WorkflowExecuteMode, +} from 'n8n-workflow'; +import { NodeConnectionTypes, UserError } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import type { CredentialsPermissionChecker } from '@/executions/pre-execution-checks'; +import type { ResumableExecution } from '@/interfaces'; +import type { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; +import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; + +const node = (id: string, name: string, type: string): INode => ({ + id, + name, + type, + typeVersion: 1, + position: [0, 0], + parameters: {}, +}); + +const MANUAL_TRIGGER = node('trigger-id', 'When clicking Execute', 'n8n-nodes-base.manualTrigger'); +const SET_NODE = node('set-id', 'Edit Fields', 'n8n-nodes-base.set'); + +function workflow(overrides: Partial = {}): IWorkflowBase { + return { + id: 'wf-1', + name: 'My workflow', + active: false, + isArchived: false, + activeVersionId: null, + createdAt: new Date(), + updatedAt: new Date(), + nodes: [MANUAL_TRIGGER, SET_NODE], + connections: { + [MANUAL_TRIGGER.name]: { + main: [[{ node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 }]], + }, + }, + settings: { engineType: 'v2' }, + ...overrides, + }; +} + +const taskData = (main: ITaskDataConnections['main']): ITaskData => ({ + startTime: 0, + executionIndex: 0, + executionTime: 0, + source: [], + data: { main }, +}); + +function runData( + overrides: Partial = {}, +): IWorkflowExecutionDataProcess { + return { + executionMode: 'manual', + workflowData: workflow(), + triggerToStartFrom: { name: MANUAL_TRIGGER.name }, + ...overrides, + }; +} + +describe('EngineV2Dispatcher', () => { + const proxy = mock(); + const credentialsPermissionChecker = mock(); + + let dispatcher: EngineV2Dispatcher; + + beforeEach(() => { + vi.clearAllMocks(); + proxy.isAvailable.mockReturnValue(true); + proxy.startExecution.mockResolvedValue({ executionId: 'dp-uuid' }); + dispatcher = new EngineV2Dispatcher(proxy, credentialsPermissionChecker); + }); + + describe('routesToEngineV2', () => { + it('routes a manual run of a workflow that opted into engine 2.0', () => { + expect(dispatcher.routesToEngineV2(runData())).toBe(true); + }); + + it.each([ + { name: 'no engineType', settings: {} }, + { name: 'engineType v1', settings: { engineType: 'v1' as const } }, + ])('does not route a workflow with $name', ({ settings }) => { + const data = runData({ workflowData: workflow({ settings }) }); + + expect(dispatcher.routesToEngineV2(data)).toBe(false); + }); + + it.each(['webhook', 'trigger', 'retry', 'chat', 'evaluation'])( + 'does not route a %s run', + (executionMode) => { + expect(dispatcher.routesToEngineV2(runData({ executionMode }))).toBe(false); + }, + ); + + it('does not route a resumed execution', () => { + const existingExecution = mock({ executionId: '42' }); + + expect(dispatcher.routesToEngineV2(runData(), existingExecution)).toBe(false); + }); + }); + + describe('start', () => { + it('returns the data plane execution id', async () => { + await expect(dispatcher.start(runData())).resolves.toBe('dp-uuid'); + + expect(proxy.startExecution).toHaveBeenCalledWith( + expect.objectContaining({ workflowId: 'wf-1', mode: 'manual' }), + ); + }); + + it('converts the workflow to a graph', async () => { + await dispatcher.start(runData()); + + const { graph } = proxy.startExecution.mock.calls[0][0]; + expect(graph.nodes).toEqual([ + { id: MANUAL_TRIGGER.id, name: MANUAL_TRIGGER.name, type: 'trigger' }, + expect.objectContaining({ id: SET_NODE.id, type: 'v1-node' }), + ]); + expect(graph.edges).toEqual([ + { from: MANUAL_TRIGGER.id, to: SET_NODE.id, outputIndex: 0, inputIndex: 0 }, + ]); + }); + + it('converts only the branch of the selected manual trigger', async () => { + const otherTrigger = node( + 'other-trigger-id', + 'When clicking Other Execute', + 'n8n-nodes-base.manualTrigger', + ); + const otherSetNode = node('other-set-id', 'Other Edit Fields', 'n8n-nodes-base.set'); + const data = runData({ + workflowData: workflow({ + nodes: [MANUAL_TRIGGER, SET_NODE, otherTrigger, otherSetNode], + connections: { + [MANUAL_TRIGGER.name]: { + main: [[{ node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 }]], + }, + [otherTrigger.name]: { + main: [ + [ + { node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 }, + { node: otherSetNode.name, type: NodeConnectionTypes.Main, index: 0 }, + ], + ], + }, + }, + }), + }); + + await dispatcher.start(data); + + const { graph } = proxy.startExecution.mock.calls[0][0]; + expect(graph.nodes).toEqual([ + { id: MANUAL_TRIGGER.id, name: MANUAL_TRIGGER.name, type: 'trigger' }, + expect.objectContaining({ id: SET_NODE.id, type: 'v1-node' }), + ]); + expect(graph.edges).toEqual([ + { from: MANUAL_TRIGGER.id, to: SET_NODE.id, outputIndex: 0, inputIndex: 0 }, + ]); + }); + + it('checks credential permissions before converting', async () => { + const failure = new UserError('Node "X" uses invalid credential'); + credentialsPermissionChecker.check.mockRejectedValueOnce(failure); + + await expect(dispatcher.start(runData())).rejects.toThrow(failure); + + expect(credentialsPermissionChecker.check).toHaveBeenCalledWith('wf-1', [ + MANUAL_TRIGGER, + SET_NODE, + ]); + expect(proxy.startExecution).not.toHaveBeenCalled(); + }); + + it('propagates a converter rejection for an unsupported trigger', async () => { + const scheduleTrigger = { ...MANUAL_TRIGGER, type: 'n8n-nodes-base.scheduleTrigger' }; + const data = runData({ + workflowData: workflow({ nodes: [scheduleTrigger, SET_NODE] }), + }); + + await expect(dispatcher.start(data)).rejects.toThrow(UnsupportedTriggerError); + expect(proxy.startExecution).not.toHaveBeenCalled(); + }); + + describe('rejections', () => { + it('reports the module being off first', async () => { + proxy.isAvailable.mockReturnValue(false); + // also unsupported, to prove the module check wins + const data = runData({ + destinationNode: { nodeName: SET_NODE.name, mode: 'inclusive' as const }, + }); + + await expect(dispatcher.start(data)).rejects.toThrow( + 'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.', + ); + }); + + it.each([ + { + name: 'a partial execution', + data: { runData: {} as IRunData }, + message: + 'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.', + }, + { + name: 'a destination node', + data: { destinationNode: { nodeName: SET_NODE.name, mode: 'inclusive' as const } }, + message: + 'Engine 2.0 cannot run a workflow up to a single node yet. Run the whole workflow instead.', + }, + { + name: 'selected start nodes', + data: { startNodes: [mock()] }, + message: + 'Engine 2.0 cannot start from selected nodes yet. Run the whole workflow instead.', + }, + { + name: 'an AI tool run', + data: { agentRequest: { query: { [SET_NODE.name]: 'do it' }, tool: { name: 'tool' } } }, + message: 'Engine 2.0 cannot run a workflow as an AI tool yet.', + }, + { + name: 'pinned data on a non-trigger node', + data: { pinData: { [SET_NODE.name]: [{ json: { pinned: true } }] } as IPinData }, + message: + 'Engine 2.0 does not support pinned data on "Edit Fields" yet. Unpin it to run this workflow.', + }, + ])('rejects $name', async ({ data, message }) => { + const attempt = dispatcher.start(runData(data)); + + await expect(attempt).rejects.toThrow(UserError); + await expect(attempt).rejects.toThrow(message); + expect(proxy.startExecution).not.toHaveBeenCalled(); + }); + }); + + describe('triggerOutputs', () => { + const startedWith = () => proxy.startExecution.mock.calls[0][0].triggerOutputs; + + it('defaults to one slot with one empty item', async () => { + await dispatcher.start(runData()); + + expect(startedWith()).toEqual([[{ json: {} }]]); + }); + + it('passes the trigger payload through', async () => { + const data = runData({ + triggerToStartFrom: { + name: MANUAL_TRIGGER.name, + data: taskData([[{ json: { from: 'trigger' } }]]), + }, + }); + + await dispatcher.start(data); + + expect(startedWith()).toEqual([[{ json: { from: 'trigger' } }]]); + }); + + it('uses pinned data on the start trigger', async () => { + const data = runData({ + pinData: { [MANUAL_TRIGGER.name]: [{ json: { from: 'pin' } }] } as IPinData, + }); + + await dispatcher.start(data); + + expect(startedWith()).toEqual([[{ json: { from: 'pin' } }]]); + }); + + it('prefers the trigger payload over pinned data for the same trigger', async () => { + const data = runData({ + triggerToStartFrom: { + name: MANUAL_TRIGGER.name, + data: taskData([[{ json: { from: 'trigger' } }]]), + }, + pinData: { [MANUAL_TRIGGER.name]: [{ json: { from: 'pin' } }] } as IPinData, + }); + + await dispatcher.start(data); + + expect(startedWith()).toEqual([[{ json: { from: 'trigger' } }]]); + }); + + it('collapses an empty slot to a dead edge', async () => { + const data = runData({ + triggerToStartFrom: { + name: MANUAL_TRIGGER.name, + data: taskData([[]]), + }, + }); + + await dispatcher.start(data); + + expect(startedWith()).toEqual([null]); + }); + + it('sends null rather than an empty array when there are no slots', async () => { + const data = runData({ + triggerToStartFrom: { + name: MANUAL_TRIGGER.name, + data: taskData([]), + }, + }); + + await dispatcher.start(data); + + expect(startedWith()).toBeNull(); + }); + }); + }); +}); diff --git a/packages/cli/src/services/engine-data-plane-proxy.service.ts b/packages/cli/src/services/engine-data-plane-proxy.service.ts index d6c9ace672e..50dae5a0cca 100644 --- a/packages/cli/src/services/engine-data-plane-proxy.service.ts +++ b/packages/cli/src/services/engine-data-plane-proxy.service.ts @@ -18,9 +18,6 @@ export interface EngineDataPlaneProvider { * The module registers itself here on init. Without the module enabled there is * no provider, and calling into the engine throws rather than degrading silently: * a dropped execution would be worse than a loud failure. - * - * TODO(CAT-2877): nothing calls this yet. The dispatch that routes on the - * per-workflow `engineType` setting lands with the parent ticket. */ @Service() export class EngineDataPlaneProxyService implements EngineDataPlaneProvider { @@ -30,6 +27,11 @@ export class EngineDataPlaneProxyService implements EngineDataPlaneProvider { this.provider = provider; } + /** Whether the `engine-v2` module is enabled and has registered itself. */ + isAvailable(): boolean { + return this.provider !== null; + } + async startExecution(request: StartExecutionRequest): Promise { if (!this.provider) { throw new UserError( diff --git a/packages/cli/src/services/engine-v2-dispatcher.service.ts b/packages/cli/src/services/engine-v2-dispatcher.service.ts new file mode 100644 index 00000000000..ededb2b0eca --- /dev/null +++ b/packages/cli/src/services/engine-v2-dispatcher.service.ts @@ -0,0 +1,174 @@ +import { Service } from '@n8n/di'; +import type { StepSlots, TriggerOutputs } from '@n8n/engine'; +import type { + INodeExecutionData, + IWorkflowBase, + IWorkflowExecutionDataProcess, +} from 'n8n-workflow'; +import { getChildNodes, NodeConnectionTypes, UserError } from 'n8n-workflow'; + +import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks'; +import type { ResumableExecution } from '@/interfaces'; +import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; + +type ToStepOutputs = (outputs: INodeExecutionData[][]) => StepSlots; + +/** v1's payload for a manual run with no trigger data: one slot, one empty item. */ +const DEFAULT_MAIN_OUTPUT: INodeExecutionData[][] = [[{ json: {} }]]; + +/** + * Routes a run to the engine 2.0 data plane and starts it there. + * + * The single dispatch point for the v2 path: {@link routesToEngineV2} decides, + * {@link start} runs. A workflow that opts into engine 2.0 never falls back to + * v1 — anything the v2 path cannot do fails with a user-facing reason instead, + * because a silent fallback would run the workflow on an engine the user did + * not pick. + * + * No control-plane execution row is created: the data plane is the source of + * truth, and the returned execution id is its UUID. + */ +@Service() +export class EngineV2Dispatcher { + constructor( + private readonly proxy: EngineDataPlaneProxyService, + private readonly credentialsPermissionChecker: CredentialsPermissionChecker, + ) {} + + /** + * Manual runs only for now: webhook (CAT-2920) and trigger (CAT-2921) entry + * paths reuse this seam later. A resume must not start a fresh data-plane + * execution, hence the `existingExecution` check. + */ + routesToEngineV2( + data: IWorkflowExecutionDataProcess, + existingExecution?: ResumableExecution, + ): boolean { + return ( + data.workflowData.settings?.engineType === 'v2' && + data.executionMode === 'manual' && + existingExecution === undefined + ); + } + + /** Returns the data plane's execution id. */ + async start(data: IWorkflowExecutionDataProcess): Promise { + this.assertSupported(data); + + const { workflowData } = data; + + await this.credentialsPermissionChecker.check(workflowData.id, workflowData.nodes); + + // Lazily imported: a top-level import would pull the v1 step executor and + // its dependencies into every n8n process, including ones with the module off. + const { V1WorkflowConverter, toStepOutputs } = await import('@n8n/node-engine-compatibility'); + + const graph = new V1WorkflowConverter().convert(this.selectTriggerSubgraph(data)); + + const { executionId } = await this.proxy.startExecution({ + workflowId: workflowData.id, + graph, + triggerOutputs: this.toTriggerOutputs(data, toStepOutputs), + mode: 'manual', + }); + + return executionId; + } + + /** Keep only the branch that starts at the trigger selected for this manual run. */ + private selectTriggerSubgraph(data: IWorkflowExecutionDataProcess): IWorkflowBase { + const { triggerToStartFrom, workflowData } = data; + if (triggerToStartFrom === undefined) return workflowData; + + const includedNodeNames = new Set([ + triggerToStartFrom.name, + ...getChildNodes( + workflowData.connections, + triggerToStartFrom.name, + NodeConnectionTypes.Main, + -1, + ), + ]); + + return { + ...workflowData, + nodes: workflowData.nodes.filter((node) => includedNodeNames.has(node.name)), + connections: Object.fromEntries( + Object.entries(workflowData.connections).filter(([sourceNodeName]) => + includedNodeNames.has(sourceNodeName), + ), + ), + }; + } + + /** + * Rejects what the v2 path cannot do yet, in the order the user should hear + * about it: the module being off comes first, so a workflow that would also + * fail conversion does not report the conversion problem and hide the real + * cause. + */ + private assertSupported(data: IWorkflowExecutionDataProcess): void { + if (!this.proxy.isAvailable()) { + throw new UserError( + 'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.', + ); + } + + if (data.runData !== undefined) { + throw new UserError( + 'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.', + ); + } + + // The engine cannot stop at a node, so ignoring this would run nodes the + // user did not ask for, with their side effects. + if (data.destinationNode !== undefined) { + throw new UserError( + 'Engine 2.0 cannot run a workflow up to a single node yet. Run the whole workflow instead.', + ); + } + + if (data.startNodes?.length) { + throw new UserError( + 'Engine 2.0 cannot start from selected nodes yet. Run the whole workflow instead.', + ); + } + + if (data.agentRequest !== undefined) { + throw new UserError('Engine 2.0 cannot run a workflow as an AI tool yet.'); + } + + const triggerName = data.triggerToStartFrom?.name; + const pinnedNode = Object.keys(data.pinData ?? {}).find((name) => name !== triggerName); + if (pinnedNode !== undefined) { + throw new UserError( + `Engine 2.0 does not support pinned data on "${pinnedNode}" yet. Unpin it to run this workflow.`, + ); + } + } + + /** + * A manual run always carries a payload. Sending none would make the engine + * record the trigger with no slots, so every successor edge reads as dead and + * the execution completes having run nothing. + */ + private toTriggerOutputs( + data: IWorkflowExecutionDataProcess, + toStepOutputs: ToStepOutputs, + ): TriggerOutputs | null { + const triggerName = data.triggerToStartFrom?.name; + // `IPinData` values are a flat item array; the Manual Trigger has one output. + const pinned = triggerName ? data.pinData?.[triggerName] : undefined; + const main = + data.triggerToStartFrom?.data?.data?.main ?? + (pinned ? [pinned] : undefined) ?? + DEFAULT_MAIN_OUTPUT; + + // v1 uses `null` for a slot it has no data for; an empty slot says the same + // thing to the engine, which `toStepOutputs` collapses back to a dead edge. + const slots = toStepOutputs(main.map((slot) => slot ?? [])); + + // The wire schema rejects an empty array; `null` is how "no slots" is sent. + return slots.length === 0 ? null : slots; + } +} diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index ebf75ed654d..7580aaf7c42 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -52,6 +52,7 @@ import { ManualExecutionService } from '@/manual-execution.service'; import { NodeTypes } from '@/node-types'; import type { ScalingService } from '@/scaling/scaling.service'; import type { Job, JobData } from '@/scaling/scaling.types'; +import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service'; @@ -93,6 +94,7 @@ export class WorkflowRunner { private readonly executionsConfig: ExecutionsConfig, private readonly storageConfig: StorageConfig, private readonly externalHooks: ExternalHooks, + private readonly engineV2Dispatcher: EngineV2Dispatcher, ) {} /** The process did error */ @@ -246,6 +248,12 @@ export class WorkflowRunner { existingExecution?: ResumableExecution, responsePromise?: IDeferredPromise, ): Promise { + // The engine 2.0 path owns the whole run: it keeps no control-plane + // execution row, so everything below here does not apply to it. + if (this.engineV2Dispatcher.routesToEngineV2(data, existingExecution)) { + return await this.engineV2Dispatcher.start(data); + } + const establishContextError = await this.establishContextForPersistence(data); // Register a new execution diff --git a/packages/cli/test/integration/workflows/workflows.controller.test.ts b/packages/cli/test/integration/workflows/workflows.controller.test.ts index 476090618a3..54b642b31c8 100644 --- a/packages/cli/test/integration/workflows/workflows.controller.test.ts +++ b/packages/cli/test/integration/workflows/workflows.controller.test.ts @@ -42,11 +42,13 @@ import { v4 as uuid } from 'uuid'; import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { CollaborationService } from '@/collaboration/collaboration.service'; import { EventService } from '@/events/event.service'; +import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; import { ProjectService } from '@/services/project.service.ee'; import { WorkflowValidationService } from '@/workflows/workflow-validation.service'; import { createFolder } from '@test-integration/db/folders'; import { saveCredential } from '../shared/db/credentials'; +import { getAllExecutions } from '../shared/db/executions'; import { createCustomRoleWithScopeSlugs, cleanupRolesAndScopes } from '../shared/db/roles'; import { assignTagToWorkflow, createTag } from '../shared/db/tags'; import { @@ -4968,6 +4970,96 @@ describe('POST /workflows/:workflowId/run', () => { 'To run the workflow manually, specify either a trigger to start from or a destination node.', ); }); + + describe('with engineType v2', () => { + const TRIGGER_NAME = 'When clicking Execute'; + const SET_NAME = 'Edit Fields'; + + const startExecution = vi.fn(); + + beforeAll(() => { + Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution }); + }); + + beforeEach(() => { + startExecution.mockResolvedValue({ executionId: 'a3c1e0f2-0000-4000-8000-000000000001' }); + }); + + const createV2Workflow = async () => + await createWorkflow( + { + nodes: [ + { + id: uuid(), + name: TRIGGER_NAME, + type: 'n8n-nodes-base.manualTrigger', + parameters: {}, + typeVersion: 1, + position: [0, 0], + }, + { + id: uuid(), + name: SET_NAME, + type: 'n8n-nodes-base.set', + parameters: {}, + typeVersion: 3.4, + position: [200, 0], + }, + ], + connections: { + [TRIGGER_NAME]: { main: [[{ node: SET_NAME, type: 'main', index: 0 }]] }, + }, + settings: { engineType: 'v2' }, + }, + owner, + ); + + test('should run on the data plane and persist no execution', async () => { + const dbWorkflow = await createV2Workflow(); + + const response = await authOwnerAgent + .post(`/workflows/${dbWorkflow.id}/run`) + .send({ triggerToStartFrom: { name: TRIGGER_NAME } }); + + expect(response.statusCode).toBe(200); + expect(response.body.data.executionId).toBe('a3c1e0f2-0000-4000-8000-000000000001'); + expect(startExecution).toHaveBeenCalledWith( + objectContaining({ + workflowId: dbWorkflow.id, + mode: 'manual', + triggerOutputs: [[{ json: {} }]], + }), + ); + + const executions = await getAllExecutions(); + expect(executions.filter((e) => e.workflowId === dbWorkflow.id)).toHaveLength(0); + }); + + test('should return 400 for a partial execution', async () => { + const dbWorkflow = await createV2Workflow(); + + const response = await authOwnerAgent.post(`/workflows/${dbWorkflow.id}/run`).send({ + destinationNode: { nodeName: SET_NAME, mode: 'inclusive' }, + runData: { + [TRIGGER_NAME]: [ + { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [], + data: { main: [[{ json: {} }]] }, + }, + ], + }, + }); + + expect(response.statusCode).toBe(400); + expect(response.body.message).toBe( + 'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.', + ); + expect(startExecution).not.toHaveBeenCalled(); + }); + }); }); describe('POST /workflows/:workflowId/archive', () => {