From dc3283da192df70bc8fe7e610b1e4401b87a0301 Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Wed, 23 Sep 2026 08:59:11 +0000 Subject: [PATCH] feat(core): Answer lastNode webhooks from execution responses (#38959) --- .../@n8n/config/src/configs/engine.config.ts | 9 + .../__tests__/engine-v2.module.test.ts | 2 + .../src/modules/engine-v2/engine-v2.module.ts | 6 +- ...ngine-v2-webhook-responder.service.test.ts | 181 ++++++++++++++++++ .../services/engine-v2-dispatcher.service.ts | 14 +- .../engine-v2-webhook-responder.service.ts | 122 ++++++++++++ .../src/services/pending-webhook-response.ts | 56 ++++++ .../__tests__/engine-v2-webhooks.test.ts | 75 ++++++++ .../__tests__/webhook-helpers.test.ts | 141 +++++++++++++- .../cli/src/webhooks/engine-v2-webhooks.ts | 69 ++++++- packages/cli/src/webhooks/webhook-helpers.ts | 61 +++++- .../integration/engine-v2-webhooks.test.ts | 17 +- packages/workflow/src/interfaces.ts | 6 + 13 files changed, 739 insertions(+), 20 deletions(-) create mode 100644 packages/cli/src/services/__tests__/engine-v2-webhook-responder.service.test.ts create mode 100644 packages/cli/src/services/engine-v2-webhook-responder.service.ts create mode 100644 packages/cli/src/services/pending-webhook-response.ts create mode 100644 packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts diff --git a/packages/@n8n/config/src/configs/engine.config.ts b/packages/@n8n/config/src/configs/engine.config.ts index e6bc03fab32..7c32bc2f714 100644 --- a/packages/@n8n/config/src/configs/engine.config.ts +++ b/packages/@n8n/config/src/configs/engine.config.ts @@ -1,6 +1,7 @@ import { z } from 'zod'; import { Config, Env } from '../decorators'; +import { positiveIntSchema } from '../schemas'; /** * Floor for the CP → DP shared secret. Kept in step with the engine's identity @@ -62,4 +63,12 @@ export class EngineConfig { /** Where the engine dials the control plane server. Defaults to loopback; set it when that is not reachable. */ @Env('N8N_ENGINE_CONTROL_PLANE_BASE_URL') controlPlaneBaseUrl: string = ''; + + /** + * How long (in ms) the control plane waits for a run to answer before it + * answers the webhook request itself. The response channel is at-most-once, + * so a lost answer must not hold the request open forever. + */ + @Env('N8N_ENGINE_WEBHOOK_RESPONSE_TIMEOUT', positiveIntSchema.max(2147483647)) + webhookResponseTimeout: number = 2 * 60 * 1000; } diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts index 012bc898420..2b8bb67cd6a 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts @@ -3,6 +3,7 @@ import { EngineConfig, ExecutionsConfig } from '@n8n/config'; import { Container } from '@n8n/di'; import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; +import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service'; import { EngineControlPlaneServer } from '../engine-control-plane-server'; import { EngineDataPlaneClient } from '../engine-data-plane-client'; @@ -25,6 +26,7 @@ describe('EngineV2Module', () => { runtime = mockInstance(EngineV2Runtime); client = mockInstance(EngineDataPlaneClient); controlPlaneServer = mockInstance(EngineControlPlaneServer); + mockInstance(EngineV2WebhookResponder); Container.set(EngineDataPlaneProxyService, new EngineDataPlaneProxyService()); module = new EngineV2Module(); diff --git a/packages/cli/src/modules/engine-v2/engine-v2.module.ts b/packages/cli/src/modules/engine-v2/engine-v2.module.ts index 75d8233ea80..3cc8321b65b 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.module.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.module.ts @@ -39,7 +39,7 @@ export class EngineV2Module implements ModuleInterface { const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js'); await Container.get(EngineControlPlaneServer).start(); - // Create both endpoints before the engine starts. A short run can answer + // Hand both endpoints over before the engine starts. A short run can answer // before `startExecution` returns, and responses are not replayed. const { InMemoryExecutionResponseChannel } = await import( './response-channel/in-memory-execution-response-channel.js' @@ -50,6 +50,9 @@ export class EngineV2Module implements ModuleInterface { const { InMemoryExecutionResponseReceiver } = await import( './response-channel/in-memory-execution-response-receiver.js' ); + const { EngineV2WebhookResponder } = await import( + '@/services/engine-v2-webhook-responder.service.js' + ); // In-memory for now because both planes share this process. Redis endpoints // can use the same response contracts when the planes run separately. const responseChannel = new InMemoryExecutionResponseChannel(); @@ -58,6 +61,7 @@ export class EngineV2Module implements ModuleInterface { responseChannel, Container.get(Logger).scoped('engine-v2'), ); + Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver); this.responseSender = responseSender; this.responseReceiver = responseReceiver; diff --git a/packages/cli/src/services/__tests__/engine-v2-webhook-responder.service.test.ts b/packages/cli/src/services/__tests__/engine-v2-webhook-responder.service.test.ts new file mode 100644 index 00000000000..e8dc9771528 --- /dev/null +++ b/packages/cli/src/services/__tests__/engine-v2-webhook-responder.service.test.ts @@ -0,0 +1,181 @@ +import type { Logger } from '@n8n/backend-common'; +import type { EngineConfig } from '@n8n/config'; +import type { ExecutionResponse } from '@n8n/engine'; +import { mock } from 'vitest-mock-extended'; + +import { createExecutionIdV2 } from '@/executions/execution-id'; +import type { ExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/execution-response-receiver'; +import { + EngineV2WebhookResponder, + MAX_PENDING_WEBHOOKS, +} from '@/services/engine-v2-webhook-responder.service'; + +const TIMEOUT_MS = 50_000; + +/** + * Stands in for the receiver. Only `src/modules/engine-v2/**` may import + * `@n8n/engine` at runtime, and the receiver's own suite covers the frame round + * trip; what matters here is what the responder does with a response. + * + * Handlers are per execution, as the real receiver's are. A response for + * another run reaches nobody. + */ +function fakeReceiver() { + const handlers = new Map void>>(); + + return { + receiver: { + receive: (executionId: string, handler: (r: ExecutionResponse) => void) => { + const forExecution = handlers.get(executionId) ?? []; + handlers.set(executionId, [...forExecution, handler]); + + return () => handlers.delete(executionId); + }, + } as unknown as ExecutionResponseReceiver, + deliver: (response: ExecutionResponse) => + handlers.get(response.executionId)?.forEach((h) => h(response)), + }; +} + +const endedResponse = (executionId: string, overrides: Record = {}) => ({ + type: 'ended' as const, + executionId, + workflowId: 'wf-1', + status: 'completed' as const, + lastStep: { + nodeId: 'a', + nodeName: 'A', + status: 'completed' as const, + outputs: [[{ json: { a: 1 } }]], + }, + ...overrides, +}); + +const newResponder = (timeoutMs = TIMEOUT_MS) => + new EngineV2WebhookResponder( + mock({ webhookResponseTimeout: timeoutMs }), + mock({ scoped: () => mock() }), + ); + +describe('EngineV2WebhookResponder', () => { + let deliver: (response: ExecutionResponse) => void; + let responder: EngineV2WebhookResponder; + + beforeEach(() => { + const fake = fakeReceiver(); + deliver = fake.deliver; + responder = newResponder(); + responder.useReceiver(fake.receiver); + }); + + it('listens under the id the run is started with', () => { + const executionId = createExecutionIdV2(); + + expect(responder.waitForResponse(executionId).executionId).toBe(executionId); + }); + + it('refuses to listen before the host hands over a receiver', () => { + expect(() => newResponder().waitForResponse(createExecutionIdV2())).toThrow( + 'without a receiver', + ); + }); + + it('refuses a run once it listens for as many as it can hold', () => { + for (let i = 0; i < MAX_PENDING_WEBHOOKS; i++) { + responder.waitForResponse(createExecutionIdV2()); + } + + // Refused before dispatch, so no run starts that nothing can answer. + expect(() => responder.waitForResponse(createExecutionIdV2())).toThrow('Try again later'); + }); + + it('listens again once an answered run releases its slot', () => { + const pending = Array.from({ length: MAX_PENDING_WEBHOOKS }, () => + responder.waitForResponse(createExecutionIdV2()), + ); + + pending[0].release(); + + expect(() => responder.waitForResponse(createExecutionIdV2())).not.toThrow(); + }); + + it('leaves an unrelated pending run unaffected', async () => { + const other = responder.waitForResponse(createExecutionIdV2()); + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver(endedResponse(pending.executionId)); + + await expect(pending.settled).resolves.toMatchObject({ status: 'completed' }); + + const stillPending = Symbol('still pending'); + await expect(Promise.race([other.settled, Promise.resolve(stillPending)])).resolves.toBe( + stillPending, + ); + }); + + it('reports the step the run ended with', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver(endedResponse(pending.executionId)); + + await expect(pending.settled).resolves.toEqual({ + status: 'completed', + lastNode: { nodeName: 'A', outputs: [[{ json: { a: 1 } }]] }, + }); + }); + + it('reports no last node when that step produced nothing', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver( + endedResponse(pending.executionId, { + lastStep: { nodeId: 'c', nodeName: 'C', status: 'skipped', outputs: null }, + }), + ); + + await expect(pending.settled).resolves.toEqual({ status: 'completed', lastNode: undefined }); + }); + + it('reports a failure with the node that caused it', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver( + endedResponse(pending.executionId, { + status: 'failed', + lastStep: { + nodeId: 'c', + nodeName: 'C', + status: 'failed', + outputs: null, + error: { name: 'NodeOperationError', message: 'it broke' }, + }, + }), + ); + + await expect(pending.settled).resolves.toEqual({ + status: 'failed', + nodeName: 'C', + error: { name: 'NodeOperationError', message: 'it broke' }, + }); + }); + + it('times out rather than waiting forever for a lost answer', async () => { + const impatient = newResponder(1); + impatient.useReceiver(fakeReceiver().receiver); + + await expect(impatient.waitForResponse(createExecutionIdV2()).settled).resolves.toEqual({ + status: 'timeout', + }); + }); + + it('drops later responses for a released run', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + pending.release(); + + deliver(endedResponse(pending.executionId)); + + await expect(Promise.race([pending.settled, Promise.resolve('still waiting')])).resolves.toBe( + 'still waiting', + ); + }); +}); diff --git a/packages/cli/src/services/engine-v2-dispatcher.service.ts b/packages/cli/src/services/engine-v2-dispatcher.service.ts index 489fa635dbe..7dfa12e898d 100644 --- a/packages/cli/src/services/engine-v2-dispatcher.service.ts +++ b/packages/cli/src/services/engine-v2-dispatcher.service.ts @@ -8,9 +8,10 @@ import type { WorkflowExecuteMode, } from 'n8n-workflow'; import { classifyTriggerIdentity, isTriggerNodeType, UserError } from 'n8n-workflow'; +import assert from 'node:assert'; import { toWorkflowDocument } from '@/executions/execution-data/types'; -import { createExecutionIdV2 } from '@/executions/execution-id'; +import { createExecutionIdV2, isExecutionIdV2 } from '@/executions/execution-id'; import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks'; import type { ResumableExecution } from '@/interfaces'; import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; @@ -82,7 +83,12 @@ export class EngineV2Dispatcher { return workflowData.settings?.engineType === 'v2' && ROUTED_MODES.has(executionMode); } - /** Returns the execution id this dispatch minted. */ + /** + * Returns the execution id this run uses. + * + * A caller that has to wait for the run's answer mints the id itself, so it + * can subscribe before the run can produce one. + */ async start(data: IWorkflowExecutionDataProcess): Promise { const trigger = this.resolveFiredTrigger(data); @@ -98,7 +104,9 @@ export class EngineV2Dispatcher { const graph = new V1WorkflowConverter().convert(workflowData, trigger.name); - const executionId = createExecutionIdV2(); + const executionId = data.engineExecutionId ?? createExecutionIdV2(); + // A caller that minted the id is waiting on that exact run. + assert(isExecutionIdV2(executionId), 'Engine 2.0 was given an id it cannot run'); // At the session cap this can evict another run's session, uncaught below. Rare; not worth fixing. this.registerPushSession(executionId, data, trigger); diff --git a/packages/cli/src/services/engine-v2-webhook-responder.service.ts b/packages/cli/src/services/engine-v2-webhook-responder.service.ts new file mode 100644 index 00000000000..6cacec910df --- /dev/null +++ b/packages/cli/src/services/engine-v2-webhook-responder.service.ts @@ -0,0 +1,122 @@ +import { Logger } from '@n8n/backend-common'; +import { EngineConfig } from '@n8n/config'; +import { Service } from '@n8n/di'; +import type { EndedMessage, ExecutionResponse } from '@n8n/engine'; +import { OperationalError, UnexpectedError } from 'n8n-workflow'; + +import type { ExecutionIdV2 } from '@/executions/execution-id'; +import type { + ExecutionResponseReceiver, + UnsubscribeExecutionResponse, +} from '@/modules/engine-v2/response-channel/execution-response-receiver'; +import { PendingWebhookResponse } from '@/services/pending-webhook-response'; + +/** A request that is still open, and the subscription that feeds its answer. */ +type PendingWebhook = { + response: PendingWebhookResponse; + unsubscribe: UnsubscribeExecutionResponse; +}; + +/** + * How many runs this replica listens for at once. Every entry has its own + * response timeout, so the map drains on its own; this only bounds it. + */ +export const MAX_PENDING_WEBHOOKS = 5000; + +/** + * A service for hooking up responses from the data plane with webhook callers + * expecting that response. + * + * `waitForResponse` should be called before starting the execution in case a + * fast execution sends a response before we're listening for it. + */ +@Service() +export class EngineV2WebhookResponder { + private receiver?: ExecutionResponseReceiver; + + private readonly pendingWebhooks = new Map(); + + constructor( + private readonly engineConfig: EngineConfig, + private readonly logger: Logger, + ) { + this.logger = this.logger.scoped('engine-v2'); + } + + /** The host calls this once with the receiver for execution responses. */ + useReceiver(receiver: ExecutionResponseReceiver): void { + if (this.receiver) { + throw new UnexpectedError('Engine 2.0 webhook response receiver is already set'); + } + this.receiver = receiver; + } + + /** + * Call this before starting the execution. Otherwise, a fast execution can + * finish before we are waiting for its response. + * + * @param executionId The caller must mint this ID, use it here, and pass it to + * `StartExecution`. + * @throws {UnexpectedError} If the execution response receiver is not set. + * @throws {OperationalError} If the service is at capacity. + */ + waitForResponse(executionId: ExecutionIdV2): PendingWebhookResponse { + const { receiver } = this; + if (!receiver) { + throw new UnexpectedError('Engine 2.0 cannot wait for a response without a receiver'); + } + + if (this.pendingWebhooks.size >= MAX_PENDING_WEBHOOKS) { + throw new OperationalError( + `Engine 2.0 already awaits ${MAX_PENDING_WEBHOOKS} webhook responses. Try again later.`, + ); + } + + const response = new PendingWebhookResponse({ + executionId, + timeoutMs: this.engineConfig.webhookResponseTimeout, + onRelease: (id) => this.release(id), + }); + const unsubscribe = receiver.receive(executionId, (received) => + this.handle(received, response), + ); + this.pendingWebhooks.set(executionId, { response, unsubscribe }); + + return response; + } + + private handle(received: ExecutionResponse, response: PendingWebhookResponse): void { + try { + this.onEnded(received, response); + } catch (error) { + this.logger.error('Failed to relay an engine 2.0 response', { + executionId: received.executionId, + type: received.type, + error, + }); + } + } + + private onEnded(received: EndedMessage, response: PendingWebhookResponse): void { + const { nodeName, outputs, error } = received.lastStep; + + if (received.status === 'failed') { + // The step that ended a failed run is the one that failed, so its name + // and error are what the caller reports. + response.resolve({ status: 'failed', nodeName, error }); + return; + } + + response.resolve({ + status: 'completed', + // A skipped or failed step carries nothing to answer with. + lastNode: outputs ? { nodeName, outputs } : undefined, + }); + } + + /** Ends the run's subscription: nothing more can arrive for it. */ + private release(executionId: string): void { + this.pendingWebhooks.get(executionId)?.unsubscribe(); + this.pendingWebhooks.delete(executionId); + } +} diff --git a/packages/cli/src/services/pending-webhook-response.ts b/packages/cli/src/services/pending-webhook-response.ts new file mode 100644 index 00000000000..8df5692297d --- /dev/null +++ b/packages/cli/src/services/pending-webhook-response.ts @@ -0,0 +1,56 @@ +import type { StepSlots } from '@n8n/engine'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; + +import type { ExecutionIdV2 } from '@/executions/execution-id'; + +/** The last node that ran, as the `lastNode` response mode answers with. */ +export type LastNode = { nodeName: string; outputs: StepSlots }; + +/** What a data-plane run produced, from the open request's point of view. */ +export type WebhookRunOutcome = + | { status: 'completed'; lastNode?: LastNode } + | { status: 'failed'; nodeName: string; error?: { name: string; message: string } } + | { status: 'timeout' }; + +export type PendingWebhookResponseOptions = { + /** The run whose answer this handle carries. */ + executionId: ExecutionIdV2; + /** How long to hold the request before it is answered without the run. */ + timeoutMs: number; + /** Called once the answer is no longer needed, so the listener can go. */ + onRelease: (executionId: string) => void; +}; + +/** + * This represents a pending webhook response. The webhook responder connects it + * to the execution responses coming from the engine. + */ +export class PendingWebhookResponse { + /** Resolves when the run ends, or when the hold runs out — whichever is first. */ + readonly settled: Promise; + + readonly executionId: ExecutionIdV2; + + private readonly onRelease: (executionId: string) => void; + + private readonly answer = createDeferredPromise(); + + /** Held so an answered run does not leave a timer behind for the whole hold. */ + private readonly timer: NodeJS.Timeout; + + constructor({ executionId, timeoutMs, onRelease }: PendingWebhookResponseOptions) { + this.executionId = executionId; + this.onRelease = onRelease; + this.settled = this.answer.promise; + this.timer = setTimeout(() => this.answer.resolve({ status: 'timeout' }), timeoutMs).unref(); + } + + resolve(outcome: WebhookRunOutcome): void { + this.answer.resolve(outcome); + } + + release(): void { + clearTimeout(this.timer); + this.onRelease(this.executionId); + } +} diff --git a/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts b/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts new file mode 100644 index 00000000000..252be4d7b17 --- /dev/null +++ b/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts @@ -0,0 +1,75 @@ +import type { StepSlots } from '@n8n/engine'; +import { WorkflowOperationError } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; +import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; +import { EngineV2PayloadGuard } from '@/services/engine-v2-payload-guard.service'; +import { EngineV2Webhooks } from '@/webhooks/engine-v2-webhooks'; + +const engineV2Webhooks = new EngineV2Webhooks( + mock(), + mock(), + mock(), +); + +describe('EngineV2Webhooks.toRun', () => { + it('converts a completed outcome with outputs', async () => { + const outputs: StepSlots = [[{ json: { answer: 42 } }], null]; + + const run = await engineV2Webhooks.toRun( + { status: 'completed', lastNode: { nodeName: 'Last node', outputs } }, + 'webhook', + ); + + expect(run).toMatchObject({ + mode: 'webhook', + status: 'success', + storedAt: 'db', + data: { + resultData: { + lastNodeExecuted: 'Last node', + runData: { + 'Last node': [ + { + executionIndex: 0, + executionStatus: 'success', + source: [], + executionTime: 0, + data: { main: [[{ json: { answer: 42 } }], []] }, + }, + ], + }, + }, + }, + }); + expect(run.startedAt).toBeInstanceOf(Date); + expect(run.data.resultData.runData['Last node'][0].startTime).toEqual(expect.any(Number)); + }); + + it('converts a failed outcome', async () => { + const run = await engineV2Webhooks.toRun( + { + status: 'failed', + nodeName: 'Broken node', + error: { name: 'Error', message: 'The node failed' }, + }, + 'webhook', + ); + + expect(run.status).toBe('error'); + expect(run.data.resultData.lastNodeExecuted).toBe('Broken node'); + expect(run.data.resultData.runData).toEqual({}); + expect(run.data.resultData.error).toBeInstanceOf(WorkflowOperationError); + expect(run.data.resultData.error?.message).toBe('The node failed'); + }); + + it('converts a completed outcome with no last node data', async () => { + const run = await engineV2Webhooks.toRun({ status: 'completed' }, 'webhook'); + + expect(run.status).toBe('success'); + expect(run.data.resultData.lastNodeExecuted).toBeUndefined(); + expect(run.data.resultData.runData).toEqual({}); + expect(run.data.resultData.error).toBeUndefined(); + }); +}); diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index 70b15a88509..95f840e88d6 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts @@ -3,6 +3,7 @@ import { mockInstance } from '@n8n/backend-test-utils'; import type { Project, User } from '@n8n/db'; import { UserRepository } from '@n8n/db'; import { Container } from '@n8n/di'; +import type { EndedMessage, ExecutionResponse } from '@n8n/engine'; import type express from 'express'; import { BinaryDataService, @@ -12,6 +13,8 @@ import { isWebhookHtmlSandboxingDisabled, } from 'n8n-core'; +import type { ExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/execution-response-receiver'; + vi.mock('n8n-core', async () => ({ ...(await vi.importActual('n8n-core')), isWebhookHtmlSandboxingDisabled: vi.fn(), @@ -58,7 +61,9 @@ import { AuthService } from '@/auth/auth.service'; import type { ResponseError } from '@/errors/response-errors/abstract/response.error'; import { EventService } from '@/events/event.service'; import { WebhookResponseRelay } from '@/scaling/webhook-response-relay'; +import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; +import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service'; import { OwnershipService } from '@/services/ownership.service'; import type { ProtectedResource } from '@/services/protected-resource.registry'; import { ProtectedResourceRegistry } from '@/services/protected-resource.registry'; @@ -1127,6 +1132,7 @@ const activeExecutions = mockInstance(ActiveExecutions); const resourceRegistry = mockInstance(ProtectedResourceRegistry); // Off by default: every block but the engine 2.0 one exercises the v1 path. const engineV2Dispatcher = mockInstance(EngineV2Dispatcher); +const engineDataPlaneProxy = mockInstance(EngineDataPlaneProxyService); const executionContextService = mockInstance(ExecutionContextService); const userRepository = mockInstance(UserRepository); mockInstance(AuthService); @@ -1944,6 +1950,8 @@ describe('executeWebhook in responseNode mode when the Respond node never runs', describe('executeWebhook on engine 2.0', () => { const errorReporter = Container.get(ErrorReporter); + /** Response handlers registered by the responder, by execution ID. */ + let dataPlane: Map void>; /** The data plane mints uuidv7 ids, not the numeric ids v1 uses. */ const ENGINE_EXECUTION_ID = '019606a1-0000-7000-8000-000000000001'; @@ -2034,7 +2042,34 @@ describe('executeWebhook on engine 2.0', () => { // The webhook path asks before it can build the run data, so it goes through // `handlesWorkflow`, not `routesToEngineV2`. engineV2Dispatcher.handlesWorkflow.mockReturnValue(true); + engineDataPlaneProxy.isAvailable.mockReturnValue(true); workflowRunner.run.mockResolvedValue(ENGINE_EXECUTION_ID); + dataPlane = new Map(); + }); + + /** Ends the run the request is waiting on, as the data plane would. */ + const answerRun = (status: EndedMessage['status'], lastStep: EndedMessage['lastStep']): void => { + const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string; + dataPlane.get(executionId)?.({ + type: 'ended', + executionId, + workflowId: WORKFLOW_ID, + status, + lastStep, + }); + }; + + beforeAll(() => { + // The host hands the responder its receiver at boot. Keeping each run's + // handler is how a test plays the data plane answering. + Container.get(EngineV2WebhookResponder).useReceiver( + mock({ + receive: vi.fn((executionId: string, handler: (r: ExecutionResponse) => void) => { + dataPlane.set(executionId, handler); + return vi.fn(); + }), + }), + ); }); describe('onReceived', () => { @@ -2065,6 +2100,93 @@ describe('executeWebhook on engine 2.0', () => { }); }); + describe('lastNode', () => { + it('dispatches the run under the id the answer is listened for on', async () => { + await startWebhook({ responseMode: 'lastNode' }); + + // The listener is created before the run starts, and the run has to use + // the id it listens under, so a fast answer is not lost. + expect(workflowRunner.run).toHaveBeenCalledTimes(1); + const dispatchedId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string; + expect(dataPlane.has(dispatchedId)).toBe(true); + }); + + it('answers with the data of the step the run ended on', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'lastNode' }); + + answerRun('completed', { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + status: 'completed', + outputs: [[{ json: { ok: true } }]], + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback).toHaveBeenCalledWith( + null, + expect.objectContaining({ body: [{ ok: true }], code: 200 }), + ); + }); + + it('answers that nothing was returned when the step produced no data', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'lastNode' }); + + answerRun('completed', { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + status: 'skipped', + outputs: null, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback.mock.calls[0]).toEqual([ + null, + { + data: { + message: 'Workflow executed successfully but the last node did not return any data', + }, + responseCode: 200, + }, + ]); + }); + + it('answers with a failure when the run ended on a failed step', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'lastNode' }); + + answerRun('failed', { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + status: 'failed', + outputs: null, + error: { name: 'NodeOperationError', message: 'it broke' }, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback.mock.calls[0]).toEqual([ + null, + { data: { message: 'Error in workflow' }, responseCode: 500 }, + ]); + }); + + it('answers with a timeout when the run does not send an ended message', async () => { + const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse'); + const { responseCallback } = await startWebhook({ responseMode: 'lastNode' }); + const pending = waitForResponse.mock.results[0]?.value; + + expect(pending).toBeDefined(); + pending?.resolve({ status: 'timeout' }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback.mock.calls[0]).toEqual([ + null, + { + data: { message: 'The workflow did not answer in time' }, + responseCode: 504, + }, + ]); + }); + }); + describe('rejections', () => { const reasonFrom = (responseCallback: ReturnType) => { const error = responseCallback.mock.calls[0][0] as ResponseError; @@ -2072,12 +2194,6 @@ describe('executeWebhook on engine 2.0', () => { }; it.each([ - { - name: 'the lastNode response mode', - options: { responseMode: 'lastNode' }, - message: - "Engine 2.0 does not support the 'lastNode' response mode yet. Respond immediately instead.", - }, { name: 'the responseNode response mode', options: { responseMode: 'responseNode' }, @@ -2142,6 +2258,19 @@ describe('executeWebhook on engine 2.0', () => { expect(workflowRunner.run).not.toHaveBeenCalled(); }); + it('answers 400, not 500, when the engine-v2 module is disabled', async () => { + engineDataPlaneProxy.isAvailable.mockReturnValue(false); + + const { responseCallback } = await startWebhook(); + + expect(reasonFrom(responseCallback)).toEqual({ + status: 400, + message: + 'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.', + }); + expect(workflowRunner.run).not.toHaveBeenCalled(); + }); + it('answers 400 for a file upload, and deletes what the node already stored', async () => { const binaryDataService = Container.get(BinaryDataService); const { responseCallback } = await startWebhook({ diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index a3012d36581..148940e437c 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -1,6 +1,8 @@ import { Service } from '@n8n/di'; import type { INode, + IRun, + IRunData, IWebhookResponseData, IWorkflowBase, WebhookResponseMode, @@ -9,16 +11,20 @@ import type { import { CHAT_TRIGGER_NODE_TYPE, classifyTriggerIdentity, + createRunExecutionData, FORM_NODE_TYPE, FORM_TRIGGER_NODE_TYPE, MICROSOFT_AGENT365_TRIGGER_NODE_TYPE, UserError, WAIT_NODE_TYPE, + WorkflowOperationError, } from 'n8n-workflow'; import { MCP_TRIGGER_NODE_TYPE } from '@/constants'; +import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; import { EngineV2PayloadGuard } from '@/services/engine-v2-payload-guard.service'; +import type { WebhookRunOutcome } from '@/services/pending-webhook-response'; /** * Trigger types the v2 path cannot serve. Each carries machinery the engine @@ -34,6 +40,13 @@ const UNSUPPORTED_TRIGGERS = new Set([ WAIT_NODE_TYPE, ]); +/** + * Response modes the v2 path serves. Each mode is added here as its support + * lands, so a mode that is not ready yet fails with a reason rather than + * answering wrongly. + */ +const SUPPORTED_RESPONSE_MODES = new Set(['onReceived', 'lastNode']); + /** What the request says about a run, before the webhook node has produced anything. */ export type EngineV2WebhookRequest = { workflowStartNode: INode; @@ -54,6 +67,7 @@ export class EngineV2Webhooks { constructor( private readonly dispatcher: EngineV2Dispatcher, private readonly payloadGuard: EngineV2PayloadGuard, + private readonly proxy: EngineDataPlaneProxyService, ) {} /** Whether this webhook run starts on the engine 2.0 data plane. */ @@ -77,6 +91,15 @@ export class EngineV2Webhooks { * Ordered so the user hears the most fundamental reason first. */ assertSupported({ workflowStartNode, responseMode, executionId }: EngineV2WebhookRequest): void { + // Checked first: `EngineV2WebhookResponder.waitForResponse` assumes the module + // registered its channel, and throws an internal error otherwise. Only a check + // that precedes that call can turn "module off" into a 400 instead of a 500. + if (!this.proxy.isAvailable()) { + throw new UserError( + 'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.', + ); + } + // A v2 run keeps no control-plane execution row, so there is nothing to resume. if (executionId !== undefined) { throw new UserError('Engine 2.0 cannot resume a waiting execution yet.'); @@ -99,14 +122,56 @@ export class EngineV2Webhooks { ); } - // TODO(CAT-4313): support `lastNode`. TODO(CAT-4079): support `responseNode`. - if (responseMode !== 'onReceived') { + // TODO(CAT-4079): Support `responseNode`. + if (!SUPPORTED_RESPONSE_MODES.has(responseMode)) { throw new UserError( `Engine 2.0 does not support the '${responseMode}' response mode yet. Respond immediately instead.`, ); } } + /** Converts the data plane's answer to the shape the v1 response path reads. */ + async toRun( + outcome: Exclude, + executionMode: WorkflowExecuteMode, + ): Promise { + const runData: IRunData = {}; + let lastNodeExecuted = outcome.status === 'failed' ? outcome.nodeName : undefined; + + if (outcome.status === 'completed' && outcome.lastNode) { + const { fromStepInputs } = await import('@n8n/node-engine-compatibility'); + lastNodeExecuted = outcome.lastNode.nodeName; + runData[lastNodeExecuted] = [ + { + startTime: Date.now(), + executionIndex: 0, + source: [], + executionTime: 0, + executionStatus: 'success', + data: { main: fromStepInputs(outcome.lastNode.outputs) }, + }, + ]; + } + + return { + mode: executionMode, + startedAt: new Date(), + status: outcome.status === 'failed' ? 'error' : 'success', + // The data plane holds the run. This object never reaches a store. + storedAt: 'db', + data: createRunExecutionData({ + resultData: { + runData, + lastNodeExecuted, + error: + outcome.status === 'failed' + ? new WorkflowOperationError(outcome.error?.message ?? 'The workflow failed') + : undefined, + }, + }), + }; + } + /** * Rejects a payload the engine cannot carry. * diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index 9f975c2d1a2..9e9819e279c 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -72,6 +72,7 @@ import { InternalServerError } from '@/errors/response-errors/internal-server.er import { NotFoundError } from '@/errors/response-errors/not-found.error'; import { UnsupportedMediaTypeError } from '@/errors/response-errors/unsupported-media-type.error'; import { EventService } from '@/events/event.service'; +import { createExecutionIdV2 } from '@/executions/execution-id'; import { parseBody } from '@/middlewares'; import { WebhookResponseRelay } from '@/scaling/webhook-response-relay'; import { @@ -81,6 +82,8 @@ import { import { OAuth2FlowProxy } from '@/services/oauth2-flow-proxy.service'; import { OwnershipService } from '@/services/ownership.service'; import { ProtectedResourceRegistry } from '@/services/protected-resource.registry'; +import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service'; +import type { PendingWebhookResponse } from '@/services/pending-webhook-response'; import { WorkflowStatisticsService } from '@/services/workflow-statistics.service'; import { WaitTracker } from '@/wait-tracker'; import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants'; @@ -835,6 +838,7 @@ export async function executeWebhook( let didSendResponse = false; /** Whether this run goes to the engine 2.0 data plane instead of the v1 path. */ let routesToEngineV2 = false; + let pendingEngineV2Response: PendingWebhookResponse | undefined; let runExecutionDataMerge = {}; const engineV2Webhooks = Container.get(EngineV2Webhooks); let cleanupMultipartFiles: (() => Promise) | undefined; @@ -1105,6 +1109,16 @@ export async function executeWebhook( didSendResponse = true; } + // Before the run, because a short workflow answers before `startExecution` + // returns and nothing replays a missed response. The id is minted here, so + // the run and the listener agree on it. + if (routesToEngineV2 && responseMode !== 'onReceived') { + const engineExecutionId = createExecutionIdV2(); + pendingEngineV2Response = + Container.get(EngineV2WebhookResponder).waitForResponse(engineExecutionId); + runData.engineExecutionId = engineExecutionId; + } + // Extract W3C trace context from webhook headers for OTEL propagation. const traceparent = req.headers.traceparent; if ( @@ -1195,14 +1209,46 @@ export async function executeWebhook( { executionId }, ); - // Engine 2.0 serves `onReceived` only, so the response is already out. Nothing - // below applies: the run has no control-plane execution to wait on. - if (routesToEngineV2) return executionId; + if (routesToEngineV2 && responseMode === 'onReceived') return executionId; - const activeExecutions = Container.get(ActiveExecutions); + /** + * A callback to handle the pending execution response from the data plane. + * Returns the run data if we get a response in time. Otherwise sends a 504 + * timeout to the webhook caller. + */ + const waitForDataPlaneRun = async (waiting: PendingWebhookResponse) => { + try { + const outcome = await waiting.settled; + if (outcome.status !== 'timeout') { + return await engineV2Webhooks.toRun(outcome, executionMode); + } - // Get a promise which resolves when the workflow did execute and send then response - const executePromise = activeExecutions.getPostExecutePromise(executionId); + Container.get(Logger).warn('No answer arrived for an engine 2.0 webhook run', { + executionId, + workflowId: workflowData.id, + }); + // The webhook node can answer before the execution starts. Do not send a + // second response if the execution response later times out. + if (!didSendResponse) { + responseCallback(null, { + data: { message: 'The workflow did not answer in time' }, + responseCode: 504, + }); + didSendResponse = true; + } + + return undefined; + } finally { + waiting.release(); + } + }; + + // Get a promise which resolves when the workflow did execute and send then response. + // Engine 2.0 keeps no control-plane execution to wait on, so its answer comes + // off the response channel, shaped like the run the handler below reads. + const executePromise = pendingEngineV2Response + ? waitForDataPlaneRun(pendingEngineV2Response) + : Container.get(ActiveExecutions).getPostExecutePromise(executionId); const { parentExecution } = runExecutionData; if (WorkflowHelpers.shouldRestartParentExecution(parentExecution)) { @@ -1332,6 +1378,9 @@ export async function executeWebhook( } return executionId; } catch (e) { + // Nothing will ever answer this one, so stop waiting for it. + pendingEngineV2Response?.release(); + let error: Error; if (e instanceof ResponseError && e.httpStatusCode < 500) { error = e; diff --git a/packages/cli/test/integration/engine-v2-webhooks.test.ts b/packages/cli/test/integration/engine-v2-webhooks.test.ts index b40dfd64fee..b9abb1b85a5 100644 --- a/packages/cli/test/integration/engine-v2-webhooks.test.ts +++ b/packages/cli/test/integration/engine-v2-webhooks.test.ts @@ -6,6 +6,7 @@ * assert what reaches the data plane. */ +import { Logger } from '@n8n/backend-common'; import { createWorkflow, mockInstance, testDb } from '@n8n/backend-test-utils'; import { GlobalConfig } from '@n8n/config'; import { UUID_V7_PATTERN } from '@n8n/constants'; @@ -17,7 +18,10 @@ import { randomUUID } from 'node:crypto'; import { agent as testAgent } from 'supertest'; import { CacheService } from '@/services/cache/cache.service'; +import { InMemoryExecutionResponseChannel } from '@/modules/engine-v2/response-channel/in-memory-execution-response-channel'; +import { InMemoryExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/in-memory-execution-response-receiver'; import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; +import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service'; import { Telemetry } from '@/telemetry'; import { WebhookServer } from '@/webhooks/webhook-server'; @@ -73,6 +77,13 @@ beforeAll(async () => { getExecution, searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }), }); + // The host hands the responder its receiver at boot (`EngineV2Module.init`). + // This test drives the webhook route directly, without the module, so it + // wires the same receiver by hand. + const responseChannel = new InMemoryExecutionResponseChannel(); + Container.get(EngineV2WebhookResponder).useReceiver( + new InMemoryExecutionResponseReceiver(responseChannel, Container.get(Logger)), + ); // `/webhook-test/*` is mounted only when a server opts into test webhooks. class EditorFacingWebhookServer extends WebhookServer { @@ -134,7 +145,7 @@ describe('webhook runs on engine 2.0', () => { test('answers 400 with the reason when the response mode is unsupported', async () => { const webhookId = randomUUID(); const trigger = webhookNode(webhookId); - trigger.parameters.responseMode = 'lastNode'; + trigger.parameters.responseMode = 'responseNode'; const workflow = await createV2Workflow(trigger); await startListening(workflow.id); @@ -144,7 +155,9 @@ describe('webhook runs on engine 2.0', () => { .send({ order: 42 }); expect(response.statusCode).toBe(400); - expect(response.body.message).toContain("does not support the 'lastNode' response mode yet"); + expect(response.body.message).toContain( + "does not support the 'responseNode' response mode yet", + ); expect(startExecution).not.toHaveBeenCalled(); }); }); diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index 69c867463d0..e62a1442437 100644 --- a/packages/workflow/src/interfaces.ts +++ b/packages/workflow/src/interfaces.ts @@ -3767,6 +3767,12 @@ export interface IWorkflowExecutionDataProcess { agentRequest?: AiAgentRequest; httpResponse?: express.Response; // Used for streaming responses streamingEnabled?: boolean; + /** + * Only engine 2.0 reads this. The caller mints the data-plane execution id + * when it has to wait for the run's answer, so it can subscribe before the + * run starts. + */ + engineExecutionId?: string; startedAt?: Date; // MCP-specific fields for queue mode support