refactor(core): Let the engine decide how a wait is implemented (no-changelog) (#39364)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts

@@ -20,6 +20,9 @@ import {
 	ExpressionError,
 	NodeConnectionTypes,
 	CONSOLE_OUTPUT_REDACTED_MESSAGE,
+	WAIT_INDEFINITELY,
+	WAIT_FOR_SUB_EXECUTION,
+	MAX_IN_PROCESS_WAIT_MS,
 } from 'n8n-workflow';
 import { mock } from 'vitest-mock-extended';
 
@@ -898,4 +901,136 @@ describe('ExecuteContext', () => {
 			});
 		});
 	});
+
+	describe('putExecutionToWait', () => {
+		const SHORT_MS = MAX_IN_PROCESS_WAIT_MS / 2;
+		const LONG_MS = MAX_IN_PROCESS_WAIT_MS * 10;
+
+		const makeWaitContext = (abortSignal?: AbortSignal) => {
+			const waitRunExecutionData = {
+				resultData: { runData: {} },
+			} as unknown as IRunExecutionData;
+			const waitAdditionalData = mock<IWorkflowExecuteAdditionalData>();
+			const context = new ExecuteContext(
+				workflow,
+				node,
+				waitAdditionalData,
+				mode,
+				waitRunExecutionData,
+				runIndex,
+				connectionInputData,
+				inputData,
+				executeData,
+				[],
+				abortSignal,
+			);
+			return { context, waitRunExecutionData, waitAdditionalData };
+		};
+
+		beforeEach(() => {
+			vi.useFakeTimers();
+		});
+
+		afterEach(() => {
+			vi.useRealTimers();
+		});
+
+		it('keeps the threshold at the value the tracker poll interval sets', () => {
+			expect(MAX_IN_PROCESS_WAIT_MS).toBe(65_000);
+		});
+
+		it('sleeps in the process when only a deadline can end a short wait', async () => {
+			const { context, waitRunExecutionData, waitAdditionalData } = makeWaitContext();
+
+			const pending = context.putExecutionToWait(new Date(Date.now() + SHORT_MS), {
+				acceptsResumeRequest: false,
+			});
+			let settled = false;
+			void pending.then(() => {
+				settled = true;
+			});
+
+			await vi.advanceTimersByTimeAsync(SHORT_MS - 1_000);
+			expect(settled).toBe(false);
+
+			await vi.advanceTimersByTimeAsync(1_000);
+			await pending;
+
+			expect(waitRunExecutionData.waitTill).toBeUndefined();
+			expect(waitAdditionalData.setExecutionStatus).not.toHaveBeenCalled();
+		});
+
+		it('suspends a short wait that a resume request can end', async () => {
+			const { context, waitRunExecutionData, waitAdditionalData } = makeWaitContext();
+			const waitTill = new Date(Date.now() + SHORT_MS);
+
+			await context.putExecutionToWait(waitTill, { acceptsResumeRequest: true });
+
+			expect(waitRunExecutionData.waitTill).toEqual(waitTill);
+			expect(waitAdditionalData.setExecutionStatus).toHaveBeenCalledWith('waiting');
+		});
+
+		it('suspends a wait that lands exactly on the threshold', async () => {
+			const { context, waitRunExecutionData, waitAdditionalData } = makeWaitContext();
+			const waitTill = new Date(Date.now() + MAX_IN_PROCESS_WAIT_MS);
+
+			await context.putExecutionToWait(waitTill, { acceptsResumeRequest: false });
+
+			expect(waitRunExecutionData.waitTill).toEqual(waitTill);
+			expect(waitAdditionalData.setExecutionStatus).toHaveBeenCalledWith('waiting');
+		});
+
+		it('suspends a long wait that only a deadline can end', async () => {
+			const { context, waitRunExecutionData, waitAdditionalData } = makeWaitContext();
+			const waitTill = new Date(Date.now() + LONG_MS);
+
+			await context.putExecutionToWait(waitTill, { acceptsResumeRequest: false });
+
+			expect(waitRunExecutionData.waitTill).toEqual(waitTill);
+			expect(waitAdditionalData.setExecutionStatus).toHaveBeenCalledWith('waiting');
+		});
+
+		it.each([
+			['WAIT_INDEFINITELY', WAIT_INDEFINITELY],
+			['WAIT_FOR_SUB_EXECUTION', WAIT_FOR_SUB_EXECUTION],
+		])('suspends on the %s sentinel and does not sleep', async (_name, sentinel) => {
+			const { context, waitRunExecutionData, waitAdditionalData } = makeWaitContext();
+
+			await context.putExecutionToWait(sentinel, { acceptsResumeRequest: false });
+
+			expect(waitRunExecutionData.waitTill).toEqual(sentinel);
+			expect(waitAdditionalData.setExecutionStatus).toHaveBeenCalledWith('waiting');
+		});
+
+		it('leaves no cancellation listener behind when a sleep ends normally', async () => {
+			const abortController = new AbortController();
+			const added = vi.spyOn(abortController.signal, 'addEventListener');
+			const removed = vi.spyOn(abortController.signal, 'removeEventListener');
+			const { context } = makeWaitContext(abortController.signal);
+
+			const pending = context.putExecutionToWait(new Date(Date.now() + SHORT_MS), {
+				acceptsResumeRequest: false,
+			});
+			await vi.advanceTimersByTimeAsync(SHORT_MS);
+			await pending;
+
+			// A workflow can reach many short waits. Each one must release its listener.
+			expect(removed.mock.calls.length).toBe(added.mock.calls.length);
+		});
+
+		it('ends a sleep early when the execution is cancelled', async () => {
+			const abortController = new AbortController();
+			const { context, waitRunExecutionData } = makeWaitContext(abortController.signal);
+
+			const pending = context.putExecutionToWait(new Date(Date.now() + SHORT_MS), {
+				acceptsResumeRequest: false,
+			});
+
+			abortController.abort();
+			await pending;
+
+			expect(vi.getTimerCount()).toBe(0);
+			expect(waitRunExecutionData.waitTill).toBeUndefined();
+		});
+	});
 });

packages/core/src/execution-engine/node-execution-context/base-execute-context.ts

@@ -1,3 +1,4 @@
+import { sleep } from '@n8n/utils/sleep';
 import type { Result } from '@n8n/utils/result';
 import get from 'lodash/get';
 import type {
@@ -36,6 +37,7 @@ import {
 	NodeHelpers,
 	NodeConnectionTypes,
 	WAIT_FOR_SUB_EXECUTION,
+	MAX_IN_PROCESS_WAIT_MS,
 	WorkflowDataProxy,
 	createEnvProviderState,
 	applyDynamicCredentialsUsage,
@@ -130,7 +132,19 @@ export class BaseExecuteContext extends NodeExecutionContext {
 		);
 	}
 
-	async putExecutionToWait(waitTill: Date): Promise<void> {
+	async putExecutionToWait(
+		waitTill: Date,
+		options?: { acceptsResumeRequest?: boolean },
+	): Promise<void> {
+		const waitMs = Math.max(waitTill.getTime() - Date.now(), 0);
+
+		// Suspending a short wait costs a write and a reload, and the tracker polls too
+		// slowly to promise an on-time resume.
+		if (options?.acceptsResumeRequest === false && waitMs < MAX_IN_PROCESS_WAIT_MS) {
+			// A cancelled execution ends the wait, and the step still returns its input.
+			return await sleep(waitMs, this.abortSignal).catch(() => undefined);
+		}
+
 		this.runExecutionData.waitTill = waitTill;
 		if (this.additionalData.setExecutionStatus) {
 			this.additionalData.setExecutionStatus('waiting');

packages/nodes-base/nodes/Wait/Wait.node.ts

@@ -591,22 +591,8 @@ export class Wait extends Webhook {
 			}
 		}
 
-		const waitValue = Math.max(waitTill.getTime() - new Date().getTime(), 0);
-
-		if (waitValue < 65000) {
-			// If wait time is shorter than 65 seconds leave execution active because
-			// we just check the database every 60 seconds.
-			return await new Promise((resolve, _reject) => {
-				const timer = setTimeout(() => resolve([context.getInputData()]), waitValue);
-				context.onExecutionCancellation(() => {
-					clearTimeout(timer);
-					resolve([context.getInputData()]);
-				});
-			});
-		}
-
-		// If longer than 65 seconds put execution to wait
-		return await this.putToWait(context, waitTill);
+		// "After Time Interval" and "At Specified Time" give the user no resume URL.
+		return await this.putToWait(context, waitTill, { acceptsResumeRequest: false });
 	}
 
 	private async configureAndPutToWait(context: IExecuteFunctions) {
@@ -637,11 +623,16 @@ export class Wait extends Webhook {
 			}
 		}
 
-		return await this.putToWait(context, waitTill);
+		// A request on the resume URL can end these modes before the deadline.
+		return await this.putToWait(context, waitTill, { acceptsResumeRequest: true });
 	}
 
-	private async putToWait(context: IExecuteFunctions, waitTill: Date) {
-		await context.putExecutionToWait(waitTill);
+	private async putToWait(
+		context: IExecuteFunctions,
+		waitTill: Date,
+		options: { acceptsResumeRequest: boolean },
+	) {
+		await context.putExecutionToWait(waitTill, options);
 		return [context.getInputData()];
 	}
 }

packages/nodes-base/nodes/Wait/test/Wait.node.test.ts

@@ -5,6 +5,7 @@ import {
 	FORM_TRIGGER_NODE_TYPE,
 	NodeOperationError,
 	UserError,
+	WAIT_INDEFINITELY,
 	type IExecuteFunctions,
 } from 'n8n-workflow';
 
@@ -65,47 +66,68 @@ describe('Execute Wait Node', () => {
 
 			if (isValid) {
 				await expect(waitNode.execute(executeFunctionsMock)).resolves.not.toThrow();
-				expect(putExecutionToWaitSpy).toHaveBeenCalledWith(expectedWaitTill);
+				expect(putExecutionToWaitSpy).toHaveBeenCalledWith(expectedWaitTill, {
+					acceptsResumeRequest: false,
+				});
 			} else {
 				await expect(waitNode.execute(executeFunctionsMock)).rejects.toThrow(NodeOperationError);
 			}
 		},
 	);
 
-	test('should resolve with input data if canceled', async () => {
+	test('hands a short time wait to core without sleeping itself', async () => {
 		const putExecutionToWaitSpy = vi.fn();
 		const waitNode = new Wait();
-
-		let cancelSignal: (() => void) | null = null;
-
 		const inputData = [{ json: { test: 'data' } }];
 
 		const executeFunctionsMock = mock<IExecuteFunctions>({
 			getNodeParameter: vi.fn().mockImplementation((paramName: string) => {
 				if (paramName === 'resume') return 'timeInterval';
 				if (paramName === 'unit') return 'seconds';
-				if (paramName === 'amount') return 60;
+				if (paramName === 'amount') return 30;
 			}),
 			getTimezone: vi.fn().mockReturnValue('UTC'),
 			putExecutionToWait: putExecutionToWaitSpy,
 			getInputData: vi.fn(() => inputData),
 			getNode: vi.fn(),
-			onExecutionCancellation: (handler) => {
-				cancelSignal = handler;
-			},
 		});
 
-		const waitPromise = waitNode.execute(executeFunctionsMock);
+		const before = Date.now();
+		await expect(waitNode.execute(executeFunctionsMock)).resolves.toEqual([inputData]);
+		const after = Date.now();
 
-		for (let index = 0; index < 20; index++) {
-			await new Promise((r) => setTimeout(r, 10));
-			if (cancelSignal !== null) break;
-		}
+		const [waitTill, options] = putExecutionToWaitSpy.mock.calls[0] as [Date, unknown];
+		expect(options).toEqual({ acceptsResumeRequest: false });
+		// Pin the deadline itself. A sentinel date is also a `Date`, and core would then
+		// suspend the execution with nothing able to resume it.
+		expect(waitTill.getTime()).toBeGreaterThanOrEqual(before + 30_000);
+		expect(waitTill.getTime()).toBeLessThanOrEqual(after + 30_000);
+		// Core owns the sleep and its cancellation handler.
+		expect(executeFunctionsMock.onExecutionCancellation).not.toHaveBeenCalled();
+	});
 
-		expect(cancelSignal).not.toBeNull();
-		cancelSignal!();
+	test('lets a resume request end a webhook wait', async () => {
+		const putExecutionToWaitSpy = vi.fn();
+		const waitNode = new Wait();
 
-		await expect(waitPromise).resolves.toEqual([inputData]);
+		const executeFunctionsMock = mock<IExecuteFunctions>({
+			getNodeParameter: vi.fn().mockImplementation((paramName: string) => {
+				if (paramName === 'resume') return 'webhook';
+				if (paramName === 'limitWaitTime') return false;
+			}),
+			getTimezone: vi.fn().mockReturnValue('UTC'),
+			evaluateExpression: vi.fn().mockReturnValue('https://n8n.test/resume-url'),
+			putExecutionToWait: putExecutionToWaitSpy,
+			getInputData: vi.fn(() => []),
+			getNode: vi.fn().mockReturnValue({ name: 'Wait' }),
+			getParentNodes: vi.fn().mockReturnValue([]),
+		});
+
+		await waitNode.execute(executeFunctionsMock);
+
+		expect(putExecutionToWaitSpy).toHaveBeenCalledWith(WAIT_INDEFINITELY, {
+			acceptsResumeRequest: true,
+		});
 	});
 
 	test('should fail the node when the form redirect response cannot be dispatched', async () => {
@@ -153,7 +175,6 @@ describe('Execute Wait Node', () => {
 				{
 					unit: 'seconds',
 					amount: 0,
-					mode: 'timeout',
 					expectedWaitTill: () => DateTime.now().toJSDate(),
 				},
 				{
@@ -173,7 +194,7 @@ describe('Execute Wait Node', () => {
 				},
 			])(
 				'Validate wait unit: $unit, amount: $amount',
-				async ({ unit, amount, expectedWaitTill, error, mode }) => {
+				async ({ unit, amount, expectedWaitTill, error }) => {
 					const putExecutionToWaitSpy = vi.fn();
 					const waitNode = new Wait();
 					const inputData = [{ json: { inputData: true } }];
@@ -190,16 +211,11 @@ describe('Execute Wait Node', () => {
 					});
 
 					if (!error) {
-						if (mode === 'timeout') {
-							// for short wait times (<65s) a simple timeout is used
-							const resultPromise = waitNode.execute(executeFunctionsMock);
-							vi.runAllTimers();
-							await expect(resultPromise).resolves.toEqual([inputData]);
-						} else {
-							// for longer wait times (>=65s) the execution is put to wait
-							await expect(waitNode.execute(executeFunctionsMock)).resolves.not.toThrow();
-							expect(putExecutionToWaitSpy).toHaveBeenCalledWith(expectedWaitTill?.());
-						}
+						// Every valid interval takes the same path. Core chooses how to wait.
+						await expect(waitNode.execute(executeFunctionsMock)).resolves.toEqual([inputData]);
+						expect(putExecutionToWaitSpy).toHaveBeenCalledWith(expectedWaitTill?.(), {
+							acceptsResumeRequest: false,
+						});
 					} else {
 						await expect(waitNode.execute(executeFunctionsMock)).rejects.toThrowError(error);
 					}

packages/workflow/src/constants.ts

@@ -8,6 +8,16 @@ export const WAIT_INDEFINITELY = new Date('3000-01-01T00:00:00.000Z');
 // A parent parked on a sub-execution gets its own sentinel so the waiting-executions sweep can select those rows by equality.
 export const WAIT_FOR_SUB_EXECUTION = new Date('2999-12-31T00:00:00.000Z');
 
+/**
+ * The longest wait that sleeps in the process instead of suspending the execution.
+ * `WaitTracker` polls every 60 seconds, so it can need a full minute to see a new row.
+ * A suspended wait shorter than that resumes late. The 5 extra seconds cover the delay
+ * until the row is written. `ExecutionRepository.getWaitingExecutions` must keep
+ * selecting rows further out than one poll interval. Change this number only with
+ * those two.
+ */
+export const MAX_IN_PROCESS_WAIT_MS = 65_000;
+
 export function isIndefiniteWait(waitTill: Date): boolean {
 	const time = waitTill.getTime();
 	return time === WAIT_INDEFINITELY.getTime() || time === WAIT_FOR_SUB_EXECUTION.getTime();

packages/workflow/src/interfaces.ts

@@ -1348,7 +1348,14 @@ export type IExecuteFunctions = ExecuteFunctions.GetNodeParameterFn &
 		getNodeInputs(): INodeInputConfiguration[];
 		getNodeOutputs(): INodeOutputConfiguration[];
 		getRuntimeCredential(alias: string): Promise<IDataObject[string] | undefined>;
-		putExecutionToWait(waitTill: Date): Promise<void>;
+		/**
+		 * Pauses the execution until `waitTill`.
+		 *
+		 * Set `acceptsResumeRequest` to `false` when only the deadline can end the wait.
+		 * The engine then keeps a short wait in the process, so the wait does not survive
+		 * a restart. Without the option the engine suspends and persists the execution.
+		 */
+		putExecutionToWait(waitTill: Date, options?: { acceptsResumeRequest?: boolean }): Promise<void>;
 		sendMessageToUI(message: any): void;
 		/** Whether the run's resolved redaction policy redacts console output for this execution's mode */
 		isConsoleOutputRedacted(): boolean;