From bbd35e281b100a36aa209518c5ab027eabf85a1f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Ovejero?= Date: Fri, 9 Oct 2026 14:23:36 +0000 Subject: [PATCH] fix(core): Keep waiting executions unchanged when a resume request fails (#40683) Co-authored-by: Danny Martini --- .../__tests__/waiting-webhooks.test.ts | 119 +++++++++- .../__tests__/webhook-helpers.test.ts | 204 ++++++++++++++++++ packages/cli/src/webhooks/waiting-webhooks.ts | 17 +- packages/cli/src/webhooks/webhook-helpers.ts | 12 +- 4 files changed, 338 insertions(+), 14 deletions(-) diff --git a/packages/cli/src/webhooks/__tests__/waiting-webhooks.test.ts b/packages/cli/src/webhooks/__tests__/waiting-webhooks.test.ts index 12fca4371fe..328191f779c 100644 --- a/packages/cli/src/webhooks/__tests__/waiting-webhooks.test.ts +++ b/packages/cli/src/webhooks/__tests__/waiting-webhooks.test.ts @@ -4,12 +4,13 @@ import type { IExecutionResponse } from '@n8n/db'; import type express from 'express'; import type { InstanceSettings } from 'n8n-core'; import { WAITING_TOKEN_QUERY_PARAM } from 'n8n-core'; -import type { INodeParameters, IWorkflowBase, Workflow } from 'n8n-workflow'; +import type { INodeParameters, IWebhookData, IWorkflowBase, Workflow } from 'n8n-workflow'; import { SEND_AND_WAIT_OPERATION } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; import { ConflictError, NotFoundError } from '@n8n/errors'; import type { ExecutionPersistence } from '@/executions/execution-persistence'; +import { WaitingForms } from '@/webhooks/waiting-forms'; import { WaitingWebhooks } from '@/webhooks/waiting-webhooks'; import * as WebhookHelpers from '@/webhooks/webhook-helpers'; import type { WebhookService } from '@/webhooks/webhook.service'; @@ -1700,7 +1701,54 @@ describe('WaitingWebhooks', () => { ); }); - it('should emit when execution successfully resumes', async () => { + it('should emit once with the time captured before webhook invocation', async () => { + const responseAt = new Date('2026-10-09T12:00:00.000Z'); + mockWebhookService.getNodeWebhooks.mockReturnValue([ + { + httpMethod: 'POST', + path: '', + webhookDescription: { + restartWebhook: true, + httpMethod: 'POST', + name: 'default', + path: '', + nodeType: undefined, + } as any, + }, + ] as any); + vi.spyOn(WebhookHelpers, 'executeWebhook').mockImplementation( + async (_w, _wd, _wfd, _wsn, _m, _pr, _red, _eid, _req, _res, callback, _dn, options) => { + vi.setSystemTime(new Date(responseAt.getTime() + 5000)); + options?.onResume?.(); + callback(null, { noWebhookResponse: true }); + return undefined; + }, + ); + + const mockReq = mock({ + params: { path: 'test-execution-id', suffix: undefined }, + method: 'POST', + }); + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(responseAt); + try { + await waitingWebhooks.executeWebhook(mockReq, createMockRes()); + } finally { + vi.useRealTimers(); + } + + expect(mockEventService.emit).toHaveBeenCalledTimes(1); + expect(mockEventService.emit).toHaveBeenCalledWith( + 'execution-resumed', + expect.objectContaining({ + executionId: 'test-execution-id', + workflowId: 'workflow-id', + responseAt, + }), + ); + }); + + it('should not emit when the resume request fails before the execution resumes', async () => { mockWebhookService.getNodeWebhooks.mockReturnValue([ { httpMethod: 'POST', @@ -1716,7 +1764,7 @@ describe('WaitingWebhooks', () => { ] as any); vi.spyOn(WebhookHelpers, 'executeWebhook').mockImplementation( async (_w, _wd, _wfd, _wsn, _m, _pr, _red, _eid, _req, _res, callback) => { - callback(null, { noWebhookResponse: true }); + callback(new Error('Workflow Webhook Error: Workflow could not be started!'), {}); return undefined; }, ); @@ -1725,11 +1773,13 @@ describe('WaitingWebhooks', () => { params: { path: 'test-execution-id', suffix: undefined }, method: 'POST', }); - await waitingWebhooks.executeWebhook(mockReq, createMockRes()); + await expect(waitingWebhooks.executeWebhook(mockReq, createMockRes())).rejects.toThrow( + 'Workflow could not be started', + ); - expect(mockEventService.emit).toHaveBeenCalledWith( + expect(mockEventService.emit).not.toHaveBeenCalledWith( 'execution-resumed', - expect.objectContaining({ executionId: 'test-execution-id', workflowId: 'workflow-id' }), + expect.anything(), ); }); @@ -1824,7 +1874,8 @@ describe('WaitingWebhooks', () => { }, ] as any); vi.spyOn(WebhookHelpers, 'executeWebhook').mockImplementation( - async (_w, _wd, _wfd, _wsn, _m, _pr, _red, _eid, _req, _res, done) => { + async (_w, _wd, _wfd, _wsn, _m, _pr, _red, _eid, _req, _res, done, _dn, options) => { + options?.onResume?.(); done(null, { noWebhookResponse: true }); return undefined; }, @@ -1841,5 +1892,59 @@ describe('WaitingWebhooks', () => { expect.anything(), ); }); + + it('should not emit for a matched form webhook', async () => { + const waitingForms = new WaitingForms( + mock(), + mock(), + executionPersistence, + mockWebhookService, + mockInstanceSettings, + mockEventService, + mockEndpointsConfig, + ); + executionPersistence.findSingleExecution.mockResolvedValue( + createMockExecution({ + nodeType: 'n8n-nodes-base.wait', + nodeName: 'WaitNode', + nodeId: 'node-id', + nodeParameters: { resume: 'form' }, + }), + ); + mockWebhookService.getNodeWebhooks.mockReturnValue([ + mock({ + httpMethod: 'POST', + path: '', + webhookDescription: { + restartWebhook: true, + httpMethod: 'POST', + name: 'default', + path: '', + nodeType: 'form', + }, + }), + ]); + vi.spyOn(WebhookHelpers, 'executeWebhook').mockImplementation( + async (_w, _wd, _wfd, _wsn, _m, _pr, _red, _eid, _req, _res, callback, _dn, options) => { + options?.onResume?.(); + callback(null, { noWebhookResponse: true }); + return undefined; + }, + ); + + await waitingForms.executeWebhook( + mock({ + params: { path: 'test-execution-id', suffix: undefined }, + method: 'POST', + }), + createMockRes(), + ); + + expect(WebhookHelpers.executeWebhook).toHaveBeenCalledTimes(1); + expect(mockEventService.emit).not.toHaveBeenCalledWith( + 'execution-resumed', + expect.anything(), + ); + }); }); }); diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index 806dd9ff3aa..935ff85a305 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts @@ -34,6 +34,7 @@ import type { IWorkflowBase, IRunExecutionData, IExecuteData, + ITaskData, IWebhookData, IWorkflowExecuteAdditionalData, CredentialCheckResult, @@ -1258,6 +1259,7 @@ describe('invokeWebhook', () => { expect(result).toEqual({ webhookResultData, runExecutionDataChanges: {}, + failed: false, }); expect(responseCallback).not.toHaveBeenCalled(); expect(responder.hasResponded).toBe(false); @@ -1299,6 +1301,7 @@ describe('invokeWebhook', () => { {}, ); expect(responder.hasResponded).toBe(true); + expect(result.failed).toBe(true); expect(result.webhookResultData).toEqual({ noWebhookResponse: true, workflowData: [[{ json: {} }]], @@ -3153,3 +3156,204 @@ describe('executeWebhook on engine v2', () => { }); }); }); + +describe('executeWebhook when the node webhook throws', () => { + const startNode = mock({ + name: 'Form Ending', + type: FORM_NODE_TYPE, + typeVersion: 2, + parameters: { operation: 'completion' }, + }); + + const runRequest = async ({ + runExecutionData, + executionId, + method = 'GET', + onResume, + webhookResult, + }: { + runExecutionData?: IRunExecutionData; + executionId?: string; + method?: 'GET' | 'POST'; + onResume?: () => void; + webhookResult?: IWebhookResponseData; + }) => { + ownershipService.getWorkflowProjectCached.mockResolvedValue( + mock({ id: 'project-1', name: 'Project 1' }), + ); + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue( + mock(), + ); + engineV2Dispatcher.handlesWorkflow.mockReturnValue(false); + if (webhookResult !== undefined) { + webhookService.runWebhook.mockResolvedValue(webhookResult); + } else { + webhookService.runWebhook.mockRejectedValue( + new Error('No binary data with field data found.'), + ); + } + workflowRunner.run.mockResolvedValue(EXECUTION_ID); + + const workflow = mock({ + id: WORKFLOW_ID, + name: 'Test Workflow', + nodes: { [startNode.name]: startNode }, + getChildNodes: vi.fn().mockReturnValue([]), + nodeTypes: { + getByNameAndVersion: vi + .fn() + .mockReturnValue(mock({ description: { name: 'form' } })), + }, + expression: { + getSimpleParameterValue: vi.fn( + (...args: Parameters) => + args[1] ?? args[5], + ), + getComplexParameterValue: vi.fn( + (...args: Parameters) => args[1], + ), + }, + }); + const responseCallback = vi.fn(); + + await executeWebhook( + workflow, + { + webhookDescription: { name: 'default', responseMode: 'onReceived' }, + workflowId: WORKFLOW_ID, + } as unknown as IWebhookData, + mock({ id: WORKFLOW_ID, name: 'Test Workflow' }), + startNode, + 'webhook', + undefined, + runExecutionData, + executionId, + mock({ method, headers: {}, contentType: undefined, body: {} }), + mock({ headersSent: false }), + responseCallback, + undefined, + { onResume }, + ); + + return { responseCallback }; + }; + + const expectErrorResponse = (responseCallback: ReturnType) => { + expect(responseCallback).toHaveBeenCalledTimes(1); + expect(responseCallback).toHaveBeenCalledWith( + expect.objectContaining({ message: 'Workflow Form Error: Workflow could not be started!' }), + {}, + ); + }; + + beforeEach(() => { + vi.restoreAllMocks(); + vi.clearAllMocks(); + }); + + it('should leave a finished execution as it was', async () => { + const runExecutionData = createRunExecutionData({ + executionData: { nodeExecutionStack: [] }, + }); + + const { responseCallback } = await runRequest({ runExecutionData, executionId: EXECUTION_ID }); + + expectErrorResponse(responseCallback); + expect(workflowRunner.run).not.toHaveBeenCalled(); + expect(runExecutionData.executionData?.nodeExecutionStack).toEqual([]); + expect(Container.get(ErrorReporter).error).toHaveBeenCalledTimes(1); + }); + + // On a POST resume, `disableNode` has already disabled the waiting node. + it.each([ + { method: 'GET' as const, disabled: false }, + { method: 'POST' as const, disabled: true }, + ])('should leave a waiting execution as it was on $method', async ({ method, disabled }) => { + const runData = { Trigger: [mock()] }; + const savedRunData = { Trigger: [...runData.Trigger] }; + const waitingNode = { ...startNode, disabled }; + const runExecutionData = createRunExecutionData({ + executionData: { + nodeExecutionStack: [ + { node: waitingNode, data: { main: [[{ json: { submitted: true } }]] }, source: null }, + ], + }, + resultData: { runData, lastNodeExecuted: 'Form Ending' }, + }); + + const onResume = vi.fn(); + const { responseCallback } = await runRequest({ + runExecutionData, + executionId: EXECUTION_ID, + method, + onResume, + }); + + expect(onResume).not.toHaveBeenCalled(); + + expectErrorResponse(responseCallback); + expect(workflowRunner.run).not.toHaveBeenCalled(); + expect(runExecutionData.resultData.runData).toBe(runData); + expect(runExecutionData.resultData.runData).toEqual(savedRunData); + expect(runExecutionData.resultData.error).toBeUndefined(); + expect(runExecutionData.executionData?.nodeExecutionStack[0].data.main).toEqual([ + [{ json: { submitted: true } }], + ]); + }); + + it('should call onResume once before a waiting execution resumes', async () => { + const runExecutionData = createRunExecutionData({ + executionData: { + nodeExecutionStack: [{ node: startNode, data: { main: [[]] }, source: null }], + }, + resultData: { runData: {}, lastNodeExecuted: 'Form Ending' }, + }); + const onResume = vi.fn(); + + await runRequest({ + runExecutionData, + executionId: EXECUTION_ID, + method: 'POST', + onResume, + webhookResult: { workflowData: [[{ json: { ok: true } }]] }, + }); + + expect(onResume).toHaveBeenCalledTimes(1); + expect(workflowRunner.run).toHaveBeenCalledTimes(1); + expect(onResume.mock.invocationCallOrder[0]).toBeLessThan( + workflowRunner.run.mock.invocationCallOrder[0], + ); + }); + + it('should not resume when the node returns no workflow data', async () => { + const runExecutionData = createRunExecutionData({ + executionData: { + nodeExecutionStack: [{ node: startNode, data: { main: [[]] }, source: null }], + }, + }); + const onResume = vi.fn(); + + await runRequest({ + runExecutionData, + executionId: EXECUTION_ID, + onResume, + webhookResult: {}, + }); + + expect(workflowRunner.run).not.toHaveBeenCalled(); + expect(onResume).not.toHaveBeenCalled(); + }); + + it('should still record a new execution as failed', async () => { + const { responseCallback } = await runRequest({}); + + expectErrorResponse(responseCallback); + expect(workflowRunner.run).toHaveBeenCalledTimes(1); + expect(workflowRunner.run.mock.calls[0][0].executionData?.resultData).toEqual( + expect.objectContaining({ + lastNodeExecuted: 'Form Ending', + error: expect.objectContaining({ message: 'No binary data with field data found.' }), + }), + ); + }); +}); diff --git a/packages/cli/src/webhooks/waiting-webhooks.ts b/packages/cli/src/webhooks/waiting-webhooks.ts index 259f662fa93..f565d904f52 100644 --- a/packages/cli/src/webhooks/waiting-webhooks.ts +++ b/packages/cli/src/webhooks/waiting-webhooks.ts @@ -337,12 +337,16 @@ export class WaitingWebhooks implements IWebhookManager { }); } - private emitExecutionResumedEvent(execution: IExecutionResponse, executionId: string) { + private emitExecutionResumedEvent( + execution: IExecutionResponse, + executionId: string, + responseAt: Date, + ) { this.eventService.emit('execution-resumed', { executionId, workflowId: execution.workflowData.id, resumeSource: 'webhook', // today, we only emit the 'execution-resumed' event for webhook wait nodes - responseAt: new Date(), + responseAt, }); } @@ -422,9 +426,10 @@ export class WaitingWebhooks implements IWebhookManager { const isWaitingForWebhook = !this.includeForms && !this.isSendAndWaitRequest(workflow.nodes, suffix); - if (isWaitingForWebhook) { - this.emitExecutionResumedEvent(execution, executionId); - } + const responseAt = new Date(); + const onResume = isWaitingForWebhook + ? () => this.emitExecutionResumedEvent(execution, executionId, responseAt) + : undefined; return await new Promise((resolve, reject) => { WebhookHelpers.executeWebhook( @@ -446,7 +451,7 @@ export class WaitingWebhooks implements IWebhookManager { resolve(data); }, undefined, - { storedAt: execution.storedAt }, + { storedAt: execution.storedAt, onResume }, ).catch(reject); // ensure the Promise settles even if executeWebhook throws }); } finally { diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index c336723056c..bd8d75384cd 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -131,6 +131,8 @@ const IMMEDIATE_RESPONSE_MODES = new Set([ interface WebhookInvocationResult { webhookResultData: IWebhookResponseData; runExecutionDataChanges: WebhookExecutionDataChanges; + /** If true, the response callback was called with an error. */ + failed: boolean; } interface WebhookExecutionDataChanges { @@ -581,7 +583,7 @@ export async function invokeWebhook({ node: workflowStartNode, }); - return { webhookResultData, runExecutionDataChanges: {} }; + return { webhookResultData, runExecutionDataChanges: {}, failed: false }; } catch (e: unknown) { const error = ensureError(e); const errorMessage = _privateGetWebhookErrorMessage(error, webhookType); @@ -613,6 +615,7 @@ export async function invokeWebhook({ noWebhookResponse: true, workflowData: [[{ json: {} }]], }, + failed: true, }; } } @@ -989,6 +992,8 @@ export async function executeWebhook( encryptedRunnerIdentity?: string; /** Store recorded on the execution being resumed. Unset for a new execution. */ storedAt?: ExecutionStorageLocation; + /** Called before WorkflowRunner.run for an existing execution. */ + onResume?: () => void; }, ): Promise { const responder = new WebhookResponder(responseCallback); @@ -1130,6 +1135,9 @@ export async function executeWebhook( webhookType: ['formTrigger', 'form'].includes(nodeType.description.name) ? 'Form' : 'Webhook', responder, }); + + if (invocationResult.failed && executionId !== undefined) return; + const { webhookResultData } = invocationResult; runExecutionDataMerge = invocationResult.runExecutionDataChanges; if (routesToEngineV2) engineV2Payload = webhookResultData.workflowData; @@ -1300,6 +1308,8 @@ export async function executeWebhook( // data plane does not accept the run. engineV2Payload = undefined; + if (executionId) options?.onResume?.(); + // Start now to run the workflow executionId = await Container.get(WorkflowRunner).run( runData,