feat(core): Add engine suspension at node boundaries for resumable executions

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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<IRun>());
+		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<IRun>());
+		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<IRun>());
+		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<IRun>());
+		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<IRun>());
+		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();
+	});
+});

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<IRun> {
 		// 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,

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. */