From dc7f336594ef025cce800a71f05d2fdf2754bc7f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mike=20Repe=C4=87?= Date: Wed, 2 Sep 2026 15:17:55 +0200 Subject: [PATCH] feat(core): Add engine suspension at node boundaries for resumable executions Co-Authored-By: Claude Fable 5 --- .../workflow-execute-suspend.test.ts | 163 ++++++++++++++++++ .../src/execution-engine/workflow-execute.ts | 36 ++++ .../run-execution-data.v1.ts | 8 + 3 files changed, 207 insertions(+) create mode 100644 packages/core/src/execution-engine/__tests__/workflow-execute-suspend.test.ts diff --git a/packages/core/src/execution-engine/__tests__/workflow-execute-suspend.test.ts b/packages/core/src/execution-engine/__tests__/workflow-execute-suspend.test.ts new file mode 100644 index 00000000000..17afcd3a3d1 --- /dev/null +++ b/packages/core/src/execution-engine/__tests__/workflow-execute-suspend.test.ts @@ -0,0 +1,163 @@ +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; +import type { IRun, IRunExecutionData } from 'n8n-workflow'; +import { createRunExecutionData, WorkflowOperationError } from 'n8n-workflow'; + +import * as Helpers from '@test/helpers'; + +import { DirectedGraph } from '../partial-execution-utils'; +import { createNodeData, toITaskData } from '../partial-execution-utils/__tests__/helpers'; +import { WorkflowExecute } from '../workflow-execute'; + +describe('WorkflowExecute suspension', () => { + const executionMode = 'trigger'; + const nodeTypes = Helpers.NodeTypes(); + + const trigger = createNodeData({ name: 'trigger', type: 'n8n-nodes-base.manualTrigger' }); + const node1 = createNodeData({ name: 'node1' }); + const node2 = createNodeData({ name: 'node2' }); + + const createWorkflow = () => + new DirectedGraph() + .addNodes(trigger, node1, node2) + .addConnections({ from: trigger, to: node1 }, { from: node1, to: node2 }) + .toWorkflow({ name: '', active: false, nodeTypes, settings: { executionOrder: 'v1' } }); + + /** Runs the workflow, requesting suspension right after `suspendAfterNode` executes. */ + const runAndSuspend = async (suspendAfterNode: string) => { + const workflow = createWorkflow(); + const additionalData = Helpers.WorkflowExecuteAdditionalData(createDeferredPromise()); + const workflowExecute = new WorkflowExecute(additionalData, executionMode); + + additionalData.hooks!.addHandler('nodeExecuteAfter', function (nodeName) { + if (nodeName === suspendAfterNode) workflowExecute.suspend(); + }); + + const run = await workflowExecute.run({ workflow, startNode: trigger }); + return { run, workflow }; + }; + + test('suspend mid-run parks the execution as waiting with an intact stack', async () => { + const { run } = await runAndSuspend('node1'); + + expect(run.status).toBe('waiting'); + expect(run.waitTill).toBeInstanceOf(Date); + expect(run.finished).not.toBe(true); + expect(run.data.resumeInstruction).toBe('run-stack-head'); + + // The next, not-yet-executed node is the stack head, untouched. + const stack = run.data.executionData!.nodeExecutionStack; + expect(stack).toHaveLength(1); + expect(stack[0].node.name).toBe('node2'); + expect(stack[0].node.disabled).toBe(false); + + // Executed nodes keep their run data; the unexecuted node has none. + expect(run.data.resultData.runData.node1).toHaveLength(1); + expect(run.data.resultData.runData.node2).toBeUndefined(); + }); + + test('resume executes the stack head normally and clears the instruction', async () => { + const { run, workflow } = await runAndSuspend('node1'); + + const additionalData = Helpers.WorkflowExecuteAdditionalData(createDeferredPromise()); + const resumed = await new WorkflowExecute( + additionalData, + executionMode, + run.data, + ).processRunExecutionData(workflow); + + expect(resumed.status).toBe('success'); + expect(resumed.finished).toBe(true); + expect(resumed.data.resumeInstruction).toBeUndefined(); + + // node2 ran normally: not disabled, exactly one run recorded. + expect(node2.disabled).toBe(false); + expect(resumed.data.resultData.runData.node2).toHaveLength(1); + // The legacy path would have popped node1's run data; it must survive. + expect(resumed.data.resultData.runData.node1).toHaveLength(1); + }); + + test('legacy waiting state (no resume instruction) keeps the skip-head behavior', async () => { + const workflow = createWorkflow(); + // A Wait-node-style parked state: the executed node pushed back onto the + // stack head, its run recorded, waitTill set, no resume instruction. + const parked: IRunExecutionData = createRunExecutionData({ + startData: {}, + resultData: { + runData: { + trigger: [toITaskData([{ data: {} }])], + node1: [toITaskData([{ data: {} }])], + }, + lastNodeExecuted: 'node1', + }, + executionData: { + contextData: {}, + nodeExecutionStack: [ + { node: { ...node1 }, data: { main: [[{ json: {} }]] }, source: null }, + ], + metadata: {}, + waitingExecution: {}, + waitingExecutionSource: {}, + }, + waitTill: new Date(), + }); + + const additionalData = Helpers.WorkflowExecuteAdditionalData(createDeferredPromise()); + const workflowExecute = new WorkflowExecute(additionalData, executionMode, parked); + const resumed = await workflowExecute.processRunExecutionData(workflow); + + expect(resumed.status).toBe('success'); + // Skip-head: the pushed-back node was disabled to pass its input through, + // and its pre-park run entry was popped. + expect(parked.executionData!.nodeExecutionStack[0]?.node.disabled ?? true).toBe(true); + expect(resumed.data.resultData.runData.node2).toHaveLength(1); + }); + + test('suspend after the last node yields a normal success', async () => { + const { run } = await runAndSuspend('node2'); + + expect(run.status).toBe('success'); + expect(run.finished).toBe(true); + expect(run.waitTill).toBeUndefined(); + expect(run.data.resumeInstruction).toBeUndefined(); + }); + + test('suspend before the run starts is a no-op', async () => { + const workflow = createWorkflow(); + const additionalData = Helpers.WorkflowExecuteAdditionalData(createDeferredPromise()); + const workflowExecute = new WorkflowExecute(additionalData, executionMode); + + workflowExecute.suspend(); // status is 'new', must not arm suspension + + const run = await workflowExecute.run({ workflow, startNode: trigger }); + + expect(run.status).toBe('success'); + expect(run.data.resumeInstruction).toBeUndefined(); + }); + + test('an error after suspension clears the resume markers', async () => { + const workflow = createWorkflow(); + const additionalData = Helpers.WorkflowExecuteAdditionalData(createDeferredPromise()); + const workflowExecute = new WorkflowExecute(additionalData, executionMode); + + // Simulate the race: suspension stamped the run data, then a cancel error + // reaches processSuccessExecution before the waiting state is persisted. + // @ts-expect-error private property + workflowExecute.status = 'running'; + workflowExecute.suspend(); + // @ts-expect-error private property + const runExecutionData: IRunExecutionData = workflowExecute.runExecutionData; + runExecutionData.waitTill = new Date(); + runExecutionData.resumeInstruction = 'run-stack-head'; + + const run = await workflowExecute.processSuccessExecution( + new Date(), + workflow, + new WorkflowOperationError('Workflow execution canceled'), + ); + + expect(run.status).toBe('canceled'); + expect(run.waitTill).toBeUndefined(); + expect(run.data.waitTill).toBeUndefined(); + expect(run.data.resumeInstruction).toBeUndefined(); + }); +}); diff --git a/packages/core/src/execution-engine/workflow-execute.ts b/packages/core/src/execution-engine/workflow-execute.ts index 969f2a0f99b..ec141889ae3 100644 --- a/packages/core/src/execution-engine/workflow-execute.ts +++ b/packages/core/src/execution-engine/workflow-execute.ts @@ -122,6 +122,19 @@ export class WorkflowExecute { private readonly abortController = new AbortController(); timedOut: boolean = false; + private suspensionRequested = false; + + /** + * Ask the engine to stop at the next node boundary and park the execution as + * 'waiting', so another process can resume it from the persisted state. The + * node currently executing runs to completion. Safe to call at any time; + * no-ops if the run finishes, waits, or is canceled first. + */ + suspend(): void { + if (this.status !== 'running') return; + this.suspensionRequested = true; + } + constructor( private readonly additionalData: IWorkflowExecuteAdditionalData, private readonly mode: WorkflowExecuteMode, @@ -1509,6 +1522,13 @@ export class WorkflowExecute { this.mode, ); + if (this.runExecutionData.resumeInstruction === 'run-stack-head') { + // The engine was suspended at a node boundary, so the stack head has + // not executed yet: run it normally instead of passing it through. + this.runExecutionData.resumeInstruction = undefined; + return; + } + const executionStackEntry = this.runExecutionData.executionData.nodeExecutionStack[0]; // Error reporting itself does not depend on this: `runNode` checks // `metadata.resumeError` before `node.disabled`, so the entry carrying the @@ -2253,6 +2273,15 @@ export class WorkflowExecute { return; } + // Suspension exits before popping, so the stack head is the next + // not-yet-executed node and the persisted state resumes by running it. + // The cancel check above must win over suspension. + if (this.suspensionRequested) { + this.runExecutionData.waitTill = new Date(); + this.runExecutionData.resumeInstruction = 'run-stack-head'; + return; + } + subNodeExecutionResults = makeEngineResponse(); let nodeSuccessData: INodeExecutionData[][] | null | undefined = null; @@ -2890,6 +2919,13 @@ export class WorkflowExecute { ): Promise { // Set status before creating fullRunData if (executionError !== undefined) { + // A cancel or error can land after a suspension already stamped the run + // data; clear the markers so the persisted state is not treated as + // resumable. A genuine Wait-node waitTill is never cleared here. + if (this.suspensionRequested) { + this.runExecutionData.waitTill = undefined; + this.runExecutionData.resumeInstruction = undefined; + } Logger.debug('Workflow execution finished with error', { error: executionError, workflowId: workflow.id, diff --git a/packages/workflow/src/run-execution-data/run-execution-data.v1.ts b/packages/workflow/src/run-execution-data/run-execution-data.v1.ts index c89b8b5089b..ab0f53c9a92 100644 --- a/packages/workflow/src/run-execution-data/run-execution-data.v1.ts +++ b/packages/workflow/src/run-execution-data/run-execution-data.v1.ts @@ -73,6 +73,14 @@ export interface IRunExecutionDataV1 { */ resumeToken?: string; waitTill?: Date; + /** + * How to treat the head of `nodeExecutionStack` when resuming a waiting + * execution. Absent (legacy): the head already executed and was pushed back + * (Wait node), so resume disables it to pass input through and pops its + * duplicate runData entry. 'run-stack-head': the engine was suspended at a + * node boundary, so the head has not executed yet and must run normally. + */ + resumeInstruction?: 'run-stack-head'; pushRef?: string; /** Data needed for a worker to run a manual execution. */