diff --git a/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts b/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts index cbc673dbdf3..fa0449fc545 100644 --- a/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts +++ b/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(); + 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(); + }); + }); }); diff --git a/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts b/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts index 1a3e7979293..21baf53a151 100644 --- a/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts +++ b/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 { + async putExecutionToWait( + waitTill: Date, + options?: { acceptsResumeRequest?: boolean }, + ): Promise { + 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'); diff --git a/packages/nodes-base/nodes/Wait/Wait.node.ts b/packages/nodes-base/nodes/Wait/Wait.node.ts index a642bbfbf67..5910e3f1cc9 100644 --- a/packages/nodes-base/nodes/Wait/Wait.node.ts +++ b/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()]; } } diff --git a/packages/nodes-base/nodes/Wait/test/Wait.node.test.ts b/packages/nodes-base/nodes/Wait/test/Wait.node.test.ts index ea3897fe95c..71b901491cc 100644 --- a/packages/nodes-base/nodes/Wait/test/Wait.node.test.ts +++ b/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({ 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({ + 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); } diff --git a/packages/workflow/src/constants.ts b/packages/workflow/src/constants.ts index 4c23a43a2c4..e99c19846d5 100644 --- a/packages/workflow/src/constants.ts +++ b/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(); diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index e62a1442437..d1fe92bce9c 100644 --- a/packages/workflow/src/interfaces.ts +++ b/packages/workflow/src/interfaces.ts @@ -1348,7 +1348,14 @@ export type IExecuteFunctions = ExecuteFunctions.GetNodeParameterFn & getNodeInputs(): INodeInputConfiguration[]; getNodeOutputs(): INodeOutputConfiguration[]; getRuntimeCredential(alias: string): Promise; - putExecutionToWait(waitTill: Date): Promise; + /** + * 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; sendMessageToUI(message: any): void; /** Whether the run's resolved redaction policy redacts console output for this execution's mode */ isConsoleOutputRedacted(): boolean;