From 95c40700905d83cc6bb398ff5bee214ad293d28b Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Thu, 24 Sep 2026 08:24:43 +0000 Subject: [PATCH] feat(core): Support the responseNode webhook response mode on engine 2.0 (no-changelog) (#39003) --- packages/@n8n/engine/package.json | 1 + .../src/dependencies/external-dependencies.ts | 6 + .../step-execution.integration.test.ts | 1 + .../__tests__/step-ready-handler.test.ts | 7 + .../src/execution/step-ready-handler.ts | 3 + packages/@n8n/engine/src/index.ts | 3 + .../execution-response-sender.ts | 14 +- .../execution-response.schema.ts | 6 + .../execution-response.types.ts | 14 ++ .../@n8n/engine/src/response-channel/index.ts | 3 + .../src/runtime/create-engine-runtime.ts | 1 + .../src/__tests__/fixtures.ts | 2 + .../src/__tests__/v1-response-hooks.test.ts | 82 +++++++++ .../src/v1-response-hooks.ts | 81 +++++++++ .../src/v1-step-executor.ts | 2 + .../in-memory-execution-response.test.ts | 13 ++ .../in-memory-execution-response-sender.ts | 52 +++++- ...ngine-v2-webhook-responder.service.test.ts | 41 +++++ .../engine-v2-webhook-responder.service.ts | 32 ++-- .../src/services/pending-webhook-response.ts | 19 ++- .../__tests__/webhook-helpers.test.ts | 94 ++++++++++- .../cli/src/webhooks/engine-v2-webhooks.ts | 9 +- packages/cli/src/webhooks/webhook-helpers.ts | 157 ++++++++++++------ .../integration/engine-v2-webhooks.test.ts | 6 +- pnpm-lock.yaml | 3 + 25 files changed, 567 insertions(+), 85 deletions(-) create mode 100644 packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts create mode 100644 packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts diff --git a/packages/@n8n/engine/package.json b/packages/@n8n/engine/package.json index 2495e5de6c8..9f4f4a68ab9 100644 --- a/packages/@n8n/engine/package.json +++ b/packages/@n8n/engine/package.json @@ -28,6 +28,7 @@ "@n8n/constants": "workspace:*", "@n8n/di": "workspace:*", "@n8n/typeorm": "workspace:*", + "@n8n/utils": "workspace:*", "express": "catalog:", "jsonwebtoken": "catalog:", "pg": "catalog:", diff --git a/packages/@n8n/engine/src/dependencies/external-dependencies.ts b/packages/@n8n/engine/src/dependencies/external-dependencies.ts index f310ccc1fc7..be492ec32eb 100644 --- a/packages/@n8n/engine/src/dependencies/external-dependencies.ts +++ b/packages/@n8n/engine/src/dependencies/external-dependencies.ts @@ -1,6 +1,7 @@ import type { CallerContext, ExecutionMode, StepSlots, WaitDeclaration } from '../execution'; import type { GraphNode } from '../graph'; import type { LifecycleEventCallback } from '../lifecycle-events'; +import type { ResponseEmitter } from '../response-channel'; /** * Host integration seam — how the engine reaches capabilities it does not own. @@ -42,6 +43,11 @@ export interface StepExecutionRequest { /** Input slots gathered from predecessor steps; slot contents are opaque. */ inputs: StepSlots; context: StepExecutionContext; + /** + * Response emitter allows the step executor to send messages to the control + * plane side via the execution response channel. + */ + respond: ResponseEmitter; } /** diff --git a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts index 2437cf556ef..74f59d7fc74 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts @@ -546,6 +546,7 @@ describe('step execution (integration)', () => { orchestrationQueue, { v1StepExecutor: { execute } }, noopLifecycleEventPublisher, + noopExecutionResponseSender, ); const executionId = generateId(); diff --git a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts index 779608182ca..8d88c472f52 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest'; import type { ExternalDependencies, IStepExecutor } from '../../dependencies'; import { deriveLoops, type WorkflowGraph } from '../../graph'; import type { LifecycleEventPublisher, LifecycleEvent } from '../../lifecycle-events'; +import { noopExecutionResponseSender, type ExecutionResponseSender } from '../../response-channel'; import type { OrchestrationMessage, WorkQueue } from '../../queue'; import type { ExecutionRecord, ExecutionStore } from '../execution-store'; import { @@ -56,6 +57,7 @@ function makeHandler( queue: WorkQueue, dependencies: ExternalDependencies, lifecycleEventPublisher: LifecycleEventPublisher = makeLifecycleEventPublisher(), + responseSender: ExecutionResponseSender = noopExecutionResponseSender, onStepSuspended?: () => void, ): StepReadyHandler { return new StepReadyHandler( @@ -64,6 +66,7 @@ function makeHandler( queue, dependencies, lifecycleEventPublisher, + responseSender, onStepSuspended, ); } @@ -170,6 +173,8 @@ describe('StepReadyHandler', () => { iteration: 0, callerContext: { hostMode: 'trigger' }, }, + // The step can answer the caller while it runs. + respond: { send: expect.any(Function) }, }); expect(stepStore.completeStep).toHaveBeenCalledWith('step-a', [[{ json: { ok: true } }]]); expect(stepStore.failStep).not.toHaveBeenCalled(); @@ -757,6 +762,7 @@ describe('StepReadyHandler waits', () => { makeQueue(), { v1StepExecutor: makeExecutor({ wait: timeWait }) }, makeLifecycleEventPublisher(), + noopExecutionResponseSender, onStepSuspended, ); @@ -775,6 +781,7 @@ describe('StepReadyHandler waits', () => { makeQueue(), { v1StepExecutor: makeExecutor({ wait: timeWait }) }, makeLifecycleEventPublisher(), + noopExecutionResponseSender, onStepSuspended, ); diff --git a/packages/@n8n/engine/src/execution/step-ready-handler.ts b/packages/@n8n/engine/src/execution/step-ready-handler.ts index 06bce1d9b4a..5c04b90837a 100644 --- a/packages/@n8n/engine/src/execution/step-ready-handler.ts +++ b/packages/@n8n/engine/src/execution/step-ready-handler.ts @@ -8,6 +8,7 @@ import { type WorkflowLoop, } from '../graph'; import type { LifecycleEventPublisher } from '../lifecycle-events'; +import type { ExecutionResponseSender } from '../response-channel'; import { runBatchStep } from './batch-step'; import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; @@ -49,6 +50,7 @@ export class StepReadyHandler { private readonly orchestrationQueue: WorkQueue, private readonly dependencies: ExternalDependencies, private readonly lifecycleEventPublisher: LifecycleEventPublisher, + private readonly responseSender: ExecutionResponseSender, /** Called once a wait is on the row, so the sweeper can re-arm on its deadline. */ private readonly onStepSuspended: () => void = () => {}, ) {} @@ -179,6 +181,7 @@ export class StepReadyHandler { iteration: step.iteration, callerContext: execution.callerContext, }, + respond: this.responseSender.emitterFor(execution.id), }); if (result.wait) validateWaitDeclaration(step.id, result.wait); diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index c1445d83dda..2ac3fb2074b 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/packages/@n8n/engine/src/index.ts @@ -37,11 +37,14 @@ export type { export { executionResponseSchema, noopExecutionResponseSender, + noopResponseEmitter, } from './response-channel'; export type { EndedMessage, ExecutionResponse, ExecutionResponseSender, + ResponseEmitter, + ResponseMessage, UndeliverableMessage, } from './response-channel'; diff --git a/packages/@n8n/engine/src/response-channel/execution-response-sender.ts b/packages/@n8n/engine/src/response-channel/execution-response-sender.ts index 7f68ebe5699..eea44d953ea 100644 --- a/packages/@n8n/engine/src/response-channel/execution-response-sender.ts +++ b/packages/@n8n/engine/src/response-channel/execution-response-sender.ts @@ -1,13 +1,21 @@ -import type { ExecutionResponse } from './execution-response.types'; +import { createResultOk, type Result } from '@n8n/utils/result'; + +import { + noopResponseEmitter, + type ExecutionResponse, + type ResponseEmitter, +} from './execution-response.types'; /** Sends responses from an execution back to its caller. */ export interface ExecutionResponseSender { - send(response: ExecutionResponse): void; + send(response: ExecutionResponse): Result; + emitterFor(executionId: string): ResponseEmitter; stop(): Promise; } /** Response sender for a host that discards execution responses. */ export const noopExecutionResponseSender: ExecutionResponseSender = Object.freeze({ - send: () => {}, + send: () => createResultOk(undefined), + emitterFor: (_executionId: string) => noopResponseEmitter, stop: async () => {}, }); diff --git a/packages/@n8n/engine/src/response-channel/execution-response.schema.ts b/packages/@n8n/engine/src/response-channel/execution-response.schema.ts index d4ca5901fef..e6540888d10 100644 --- a/packages/@n8n/engine/src/response-channel/execution-response.schema.ts +++ b/packages/@n8n/engine/src/response-channel/execution-response.schema.ts @@ -18,6 +18,12 @@ export const executionResponseSchema = z.discriminatedUnion('type', [ message: z.string().min(1), }), }), + z.object({ + type: z.literal('response'), + executionId: z.string().min(1), + // A missing or undefined payload means an empty response. + payload: jsonValueSchema.optional(), + }), z.object({ type: z.literal('ended'), executionId: z.string().min(1), diff --git a/packages/@n8n/engine/src/response-channel/execution-response.types.ts b/packages/@n8n/engine/src/response-channel/execution-response.types.ts index 859179deeb5..3f072131ae1 100644 --- a/packages/@n8n/engine/src/response-channel/execution-response.types.ts +++ b/packages/@n8n/engine/src/response-channel/execution-response.types.ts @@ -1,3 +1,4 @@ +import { createResultOk, type Result } from '@n8n/utils/result'; import type { z } from 'zod'; import type { executionResponseSchema } from './execution-response.schema'; @@ -28,3 +29,16 @@ export type UndeliverableMessage = Extract; + +/** A message that a step produced a response */ +export type ResponseMessage = Extract; + +/** An emitter that allows the step execution to produce a response */ +export interface ResponseEmitter { + send(payload: unknown): Result; +} + +/** For a step whose responses nobody wants. */ +export const noopResponseEmitter: ResponseEmitter = Object.freeze({ + send: () => createResultOk(undefined), +}); diff --git a/packages/@n8n/engine/src/response-channel/index.ts b/packages/@n8n/engine/src/response-channel/index.ts index 6538e1d62ab..d08c814cbc4 100644 --- a/packages/@n8n/engine/src/response-channel/index.ts +++ b/packages/@n8n/engine/src/response-channel/index.ts @@ -1,8 +1,11 @@ export { noopExecutionResponseSender } from './execution-response-sender'; export type { ExecutionResponseSender } from './execution-response-sender'; export { executionResponseSchema } from './execution-response.schema'; +export { noopResponseEmitter } from './execution-response.types'; export type { EndedMessage, ExecutionResponse, + ResponseEmitter, + ResponseMessage, UndeliverableMessage, } from './execution-response.types'; diff --git a/packages/@n8n/engine/src/runtime/create-engine-runtime.ts b/packages/@n8n/engine/src/runtime/create-engine-runtime.ts index ee0f08f64a4..fbf0b54b40f 100644 --- a/packages/@n8n/engine/src/runtime/create-engine-runtime.ts +++ b/packages/@n8n/engine/src/runtime/create-engine-runtime.ts @@ -114,6 +114,7 @@ export function createEngineRuntime({ orchestrationQueue, dependencies, lifecycleEventPublisher, + responseSender, // A deadline set after the sweeper armed would otherwise wait for its next pass. () => waitSweeper.noteSuspended(), ), diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts index cf0bf50802a..4b401655a98 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/fixtures.ts @@ -1,4 +1,5 @@ import type { JsonObject, StepExecutionRequest, StepSlots, WorkflowGraph } from '@n8n/engine'; +import { noopResponseEmitter } from '@n8n/engine'; import type { ExecuteContext } from 'n8n-core'; import { UnrecognizedNodeTypeError } from 'n8n-core'; import { NoOp } from 'n8n-nodes-base/nodes/NoOp/NoOp.node'; @@ -261,6 +262,7 @@ export const stepRequest = ( iteration: 0, callerContext: { hostMode: 'manual' }, }, + respond: noopResponseEmitter, }); export const testStepExecutor = ( diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts new file mode 100644 index 00000000000..1250293ac77 --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts @@ -0,0 +1,82 @@ +import type { ResponseEmitter, StepExecutionRequest } from '@n8n/engine'; +import { ExecutionLifecycleHooks } from 'n8n-core'; +import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow'; +import { describe, expect, it, vi } from 'vitest'; + +import { attachResponseHooks } from '../v1-response-hooks'; + +const newRequest = () => { + const respond: ResponseEmitter = { + send: vi.fn(() => ({ ok: true as const, result: undefined })), + }; + const request = { + context: { + executionId: 'exec-1', + stepId: 'step-1', + workflowId: 'wf-1', + mode: 'production', + iteration: 0, + callerContext: { hostMode: 'webhook' }, + }, + respond, + } as unknown as StepExecutionRequest; + + return { request, respond }; +}; + +const newAdditionalData = () => ({}) as IWorkflowExecuteAdditionalData; + +describe('attachResponseHooks', () => { + it('preserves hooks supplied by the host', async () => { + const { request, respond } = newRequest(); + const existingHandler = vi.fn(); + const hooks = new ExecutionLifecycleHooks('webhook', 'exec-1', {} as IWorkflowBase); + hooks.addHandler('sendChunk', existingHandler); + const additionalData = { ...newAdditionalData(), hooks }; + const response = { body: { ok: true }, statusCode: 200 }; + + attachResponseHooks(additionalData, request); + await additionalData.hooks.runHook('sendResponse', [response]); + + expect(additionalData.hooks).toBe(hooks); + expect(additionalData.hooks.handlers.sendChunk).toEqual([existingHandler]); + expect(respond.send).toHaveBeenCalledWith(response); + }); + + it('sends what the Respond node produced to the channel', async () => { + const { request, respond } = newRequest(); + const additionalData = newAdditionalData(); + + attachResponseHooks(additionalData, request); + await additionalData.hooks?.runHook('sendResponse', [{ body: { ok: true }, statusCode: 200 }]); + + expect(respond.send).toHaveBeenCalledWith({ body: { ok: true }, statusCode: 200 }); + }); + + it('surfaces errors from the response channel', async () => { + const { request, respond } = newRequest(); + const error = new Error('Response failed'); + vi.mocked(respond.send).mockReturnValue({ ok: false, error }); + const additionalData = newAdditionalData(); + + attachResponseHooks(additionalData, request); + await expect( + additionalData.hooks?.runHook('sendResponse', [{ body: { ok: true }, statusCode: 200 }]), + ).rejects.toBe(error); + expect(respond.send).toHaveBeenCalledOnce(); + }); + + it.each([ + ['a buffer body', { body: Buffer.from('hi') }], + ['a stream body', { body: { pipe: () => {} } }], + ['a binary reference', { body: { binaryData: { id: 'file-1' } } }], + ])('refuses %s, which has no JSON form', async (_name, response) => { + const { request } = newRequest(); + const additionalData = newAdditionalData(); + attachResponseHooks(additionalData, request); + + await expect(additionalData.hooks?.runHook('sendResponse', [response])).rejects.toThrow( + /Engine 2.0 cannot/, + ); + }); +}); diff --git a/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts b/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts new file mode 100644 index 00000000000..4421d768952 --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts @@ -0,0 +1,81 @@ +import type { JsonValue, StepExecutionRequest } from '@n8n/engine'; +import { ExecutionLifecycleHooks } from 'n8n-core'; +import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow'; +import { UserError } from 'n8n-workflow'; + +import { toV1ExecuteMode } from './v1-adapters'; + +/** + * `ExecutionLifecycleHooks` takes a workflow, but `sendResponse` and `sendChunk` + * never read one, and this layer holds a `Workflow` instance rather than the + * `IWorkflowBase` the constructor wants. + */ +function stubWorkflow(id: string): IWorkflowBase { + const now = new Date(); + return { + id, + name: '', + active: false, + isArchived: false, + activeVersionId: null, + createdAt: now, + updatedAt: now, + nodes: [], + connections: {}, + }; +} + +/** + * Glues the v1 `sendResponse` hook to the v2 execution response channel. + * Allows the node execution to send a response to the control plane. + */ +export function attachResponseHooks( + additionalData: IWorkflowExecuteAdditionalData, + request: StepExecutionRequest, +): void { + const { context, respond } = request; + additionalData.hooks ??= new ExecutionLifecycleHooks( + toV1ExecuteMode(context), + context.executionId, + stubWorkflow(context.workflowId), + ); + + additionalData.hooks.addHandler('sendResponse', (response) => { + const result = respond.send(toJsonPayload(response)); + if (!result.ok) { + // Fail the node so the caller can see the response error. + throw result.error; + } + }); +} + +/** + * A response crosses a process boundary as JSON, so anything without a JSON + * form fails here rather than arriving silently mangled. + */ +function toJsonPayload(value: unknown): JsonValue { + assertCarriable(value); + + // Match Engine 1 queue mode. Nested Buffers and streams are not supported. + if (typeof value === 'object' && value !== null && 'body' in value) { + assertCarriable(value.body); + } + + return value as JsonValue; +} + +function assertCarriable(value: unknown): void { + if (Buffer.isBuffer(value)) { + throw new UserError('Engine 2.0 cannot send a binary webhook response yet.'); + } + + if (typeof value === 'object' && value !== null) { + if (typeof (value as { pipe?: unknown }).pipe === 'function') { + throw new UserError('Engine 2.0 cannot stream a webhook response body yet.'); + } + + if ('binaryData' in value) { + throw new UserError('Engine 2.0 cannot send a binary webhook response yet.'); + } + } +} diff --git a/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts b/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts index 7b346aa839b..39f5150dbba 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts @@ -17,6 +17,7 @@ import { } from './errors'; import { isV1NodeStepConfig } from './guards'; import { fromStepInputs, toStepOutputs } from './io'; +import { attachResponseHooks } from './v1-response-hooks'; import type { ExecutableNodeType, NodeRunResult, @@ -61,6 +62,7 @@ export class V1StepExecutor implements IStepExecutor { const additionalData = await this.deps.additionalDataFactory( toAdditionalDataContext(request.context), ); + attachResponseHooks(additionalData, request); const context = toV1ExecuteContext({ node, diff --git a/packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts b/packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts index a5627d4fe26..ab69512723e 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts @@ -69,6 +69,19 @@ describe('in-memory execution responses', () => { ); }); + it('stamps the execution ID on a response from a step', () => { + const channel = new InMemoryExecutionResponseChannel(); + const publish = vi.spyOn(channel, 'publish'); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); + + sender.emitterFor('exec-1').send({ ok: true }); + + expect(publish).toHaveBeenCalledExactlyOnceWith( + 'exec-1', + JSON.stringify({ type: 'response', executionId: 'exec-1', payload: { ok: true } }), + ); + }); + it('validates a response before delivering it', () => { const channel = new InMemoryExecutionResponseChannel(); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); diff --git a/packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts b/packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts index 75677236e09..f45f3ac40c3 100644 --- a/packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts +++ b/packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts @@ -1,12 +1,21 @@ import type { Logger } from '@n8n/backend-common'; -import type { ExecutionResponse, ExecutionResponseSender, UndeliverableMessage } from '@n8n/engine'; -import { toResult } from '@n8n/utils/result'; +import type { + ExecutionResponse, + ExecutionResponseSender, + JsonValue, + ResponseEmitter, + UndeliverableMessage, +} from '@n8n/engine'; +import { createResultError, createResultOk, toResult, type Result } from '@n8n/utils/result'; +import { UnexpectedError } from 'n8n-workflow'; import type { InMemoryExecutionResponseChannel } from './in-memory-execution-response-channel'; /** A response channel must be able to carry every frame this class produces. */ const MAX_FRAME_BYTES = 5 * 1024 * 1024; +type FrameResult = { ok: true; frame: string } | { ok: false; frame: string; error: Error }; + export class InMemoryExecutionResponseSender implements ExecutionResponseSender { private stopped = false; @@ -16,17 +25,29 @@ export class InMemoryExecutionResponseSender implements ExecutionResponseSender private readonly maxFrameBytes: number = MAX_FRAME_BYTES, ) {} - send(response: ExecutionResponse): void { - if (this.stopped) return; + send(response: ExecutionResponse): Result { + if (this.stopped) { + return createResultError(new UnexpectedError('The execution response sender has stopped.')); + } - this.channel.publish(response.executionId, this.toFrame(response)); + const frameResult = this.toFrame(response); + this.channel.publish(response.executionId, frameResult.frame); + + return frameResult.ok ? createResultOk(undefined) : createResultError(frameResult.error); + } + + /** Gives one step a response sender without exposing execution routing. */ + emitterFor(executionId: string): ResponseEmitter { + return { + send: (payload: JsonValue) => this.send({ type: 'response', executionId, payload }), + }; } async stop(): Promise { this.stopped = true; } - private toFrame(response: ExecutionResponse): string { + private toFrame(response: ExecutionResponse): FrameResult { const serialized = toResult(() => JSON.stringify(response)); if (!serialized.ok) { this.logger.warn('Could not serialize an execution response', { @@ -51,14 +72,27 @@ export class InMemoryExecutionResponseSender implements ExecutionResponseSender }); } - return serialized.result; + return { ok: true, frame: serialized.result }; } - private undeliverableFrame(executionId: string, error: UndeliverableMessage['error']): string { - return JSON.stringify({ + private undeliverableFrame( + executionId: string, + error: UndeliverableMessage['error'], + ): FrameResult { + const frame = JSON.stringify({ type: 'undeliverable', executionId, error, } satisfies UndeliverableMessage); + + return { + ok: false, + frame, + error: new UnexpectedError(error.message, { + extra: { + code: error.code, + }, + }), + }; } } 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 index a39f2387b26..fb951bf9903 100644 --- 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 @@ -136,6 +136,47 @@ describe('EngineV2WebhookResponder', () => { await expect(pending.settled).resolves.toEqual({ status: 'completed', lastNode: undefined }); }); + it('reports the response produced by the Respond node', async () => { + const pending = responder.waitForResponse(createExecutionIdV2(), true); + + deliver({ + type: 'response', + executionId: pending.executionId, + payload: { body: { ok: true }, headers: {}, statusCode: 200 }, + }); + + await expect(pending.settled).resolves.toEqual({ + status: 'response', + response: { body: { ok: true }, headers: {}, statusCode: 200 }, + }); + }); + + it('keeps the first terminal outcome', async () => { + const pending = responder.waitForResponse(createExecutionIdV2(), true); + + deliver({ + type: 'response', + executionId: pending.executionId, + payload: { body: { ok: true }, headers: {}, statusCode: 200 }, + }); + deliver(endedResponse(pending.executionId)); + + await expect(pending.settled).resolves.toMatchObject({ status: 'response' }); + }); + + it('ignores a Respond node result when the response mode waits for the last node', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver({ + type: 'response', + executionId: pending.executionId, + payload: { body: { ignored: true }, headers: {}, statusCode: 200 }, + }); + deliver(endedResponse(pending.executionId)); + + await expect(pending.settled).resolves.toMatchObject({ status: 'completed' }); + }); + it('reports a failure with the node that caused it', async () => { const pending = responder.waitForResponse(createExecutionIdV2()); diff --git a/packages/cli/src/services/engine-v2-webhook-responder.service.ts b/packages/cli/src/services/engine-v2-webhook-responder.service.ts index abe12dda07b..1cacfee3a34 100644 --- a/packages/cli/src/services/engine-v2-webhook-responder.service.ts +++ b/packages/cli/src/services/engine-v2-webhook-responder.service.ts @@ -60,7 +60,7 @@ export class EngineV2WebhookResponder { * @throws {UnexpectedError} If the execution response receiver is not set. * @throws {OperationalError} If the service is at capacity. */ - waitForResponse(executionId: ExecutionIdV2): PendingWebhookResponse { + waitForResponse(executionId: ExecutionIdV2, acceptsResponse = false): PendingWebhookResponse { const { receiver } = this; if (!receiver) { throw new UnexpectedError('Engine 2.0 cannot wait for a response without a receiver'); @@ -74,6 +74,7 @@ export class EngineV2WebhookResponder { const response = new PendingWebhookResponse({ executionId, + acceptsResponse, timeoutMs: this.engineConfig.webhookResponseTimeout, onRelease: (id) => this.release(id), }); @@ -87,15 +88,7 @@ export class EngineV2WebhookResponder { private handle(received: ExecutionResponse, response: PendingWebhookResponse): void { try { - if (received.type === 'undeliverable') { - response.resolve({ - status: 'undeliverable', - error: { name: received.error.code, message: received.error.message }, - }); - return; - } - - this.onEnded(received, response); + this.route(received, response); } catch (error) { this.logger.error('Failed to relay an engine 2.0 response', { executionId: received.executionId, @@ -105,6 +98,25 @@ export class EngineV2WebhookResponder { } } + private route(received: ExecutionResponse, response: PendingWebhookResponse): void { + switch (received.type) { + case 'undeliverable': + response.resolve({ + status: 'undeliverable', + error: { name: received.error.code, message: received.error.message }, + }); + return; + + case 'response': + response.resolveResponse(received.payload); + return; + + case 'ended': + this.onEnded(received, response); + return; + } + } + private onEnded(received: EndedMessage, response: PendingWebhookResponse): void { const { nodeName, outputs, error } = received.lastStep; diff --git a/packages/cli/src/services/pending-webhook-response.ts b/packages/cli/src/services/pending-webhook-response.ts index f5e6dc69a9e..fde9cd5e1c8 100644 --- a/packages/cli/src/services/pending-webhook-response.ts +++ b/packages/cli/src/services/pending-webhook-response.ts @@ -8,6 +8,7 @@ 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: 'response'; response: unknown } | { status: 'completed'; lastNode?: LastNode } | { status: 'failed'; nodeName: string; error?: { name: string; message: string } } /** The execution response could not be produced or delivered. */ @@ -17,6 +18,8 @@ export type WebhookRunOutcome = export type PendingWebhookResponseOptions = { /** The run whose answer this handle carries. */ executionId: ExecutionIdV2; + /** Whether a Respond to Webhook result answers this request. */ + acceptsResponse: boolean; /** 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. */ @@ -28,11 +31,13 @@ export type PendingWebhookResponseOptions = { * 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. */ + /** Resolves when the request gets an answer, or when the hold runs out. */ readonly settled: Promise; readonly executionId: ExecutionIdV2; + private readonly acceptsResponse: boolean; + private readonly onRelease: (executionId: string) => void; private readonly answer = createDeferredPromise(); @@ -40,8 +45,14 @@ export class PendingWebhookResponse { /** 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) { + constructor({ + executionId, + acceptsResponse, + timeoutMs, + onRelease, + }: PendingWebhookResponseOptions) { this.executionId = executionId; + this.acceptsResponse = acceptsResponse; this.onRelease = onRelease; this.settled = this.answer.promise; this.timer = setTimeout(() => this.answer.resolve({ status: 'timeout' }), timeoutMs).unref(); @@ -51,6 +62,10 @@ export class PendingWebhookResponse { this.answer.resolve(outcome); } + resolveResponse(response: unknown): void { + if (this.acceptsResponse) this.answer.resolve({ status: 'response', response }); + } + release(): void { clearTimeout(this.timer); this.onRelease(this.executionId); diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index cbdb5998a38..f7e5304a784 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts @@ -2356,6 +2356,94 @@ describe('executeWebhook on engine 2.0', () => { }); }); + describe('responseNode', () => { + it('answers with the response published by the data plane', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'responseNode' }); + const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string; + + dataPlane.get(executionId)?.({ + type: 'response', + executionId, + payload: { body: { ok: true }, headers: {}, statusCode: 200 }, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(workflowRunner.run).toHaveBeenCalledTimes(1); + expect(responseCallback).toHaveBeenCalledWith(null, { + data: { ok: true }, + headers: {}, + responseCode: 200, + }); + }); + + it('answers with an empty body when the response has no body key', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'responseNode' }); + const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string; + + dataPlane.get(executionId)?.({ + type: 'response', + executionId, + payload: { headers: { location: 'https://example.com' }, statusCode: 307 }, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback).toHaveBeenCalledWith(null, { + data: undefined, + headers: { location: 'https://example.com' }, + responseCode: 307, + }); + }); + + it('answers with an empty body when the Respond node never runs', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'responseNode' }); + + answerRun('completed', { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + status: 'completed', + outputs: [[{ json: { ignored: true } }]], + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback).toHaveBeenCalledWith(null, { + data: undefined, + responseCode: 200, + }); + }); + + it('answers with the channel error when the response is undeliverable', async () => { + const { responseCallback } = await startWebhook({ responseMode: 'responseNode' }); + const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string; + + dataPlane.get(executionId)?.({ + type: 'undeliverable', + executionId, + error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback).toHaveBeenCalledWith(null, { + data: { message: 'The response is too large.' }, + responseCode: 500, + }); + }); + + it('answers with a timeout when no terminal outcome arrives', async () => { + const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse'); + const { responseCallback } = await startWebhook({ responseMode: 'responseNode' }); + const pending = waitForResponse.mock.results[0]?.value; + + expect(pending).toBeDefined(); + pending?.resolve({ status: 'timeout' }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback).toHaveBeenCalledWith(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; @@ -2363,12 +2451,6 @@ describe('executeWebhook on engine 2.0', () => { }; it.each([ - { - name: 'the responseNode response mode', - options: { responseMode: 'responseNode' }, - message: - "Engine 2.0 does not support the 'responseNode' response mode yet. Respond immediately instead.", - }, { name: 'the streaming response mode', options: { responseMode: 'streaming' }, diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index e340204123c..d406e2f4bac 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -45,7 +45,11 @@ const UNSUPPORTED_TRIGGERS = new Set([ * 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']); +const SUPPORTED_RESPONSE_MODES = new Set([ + 'onReceived', + 'lastNode', + 'responseNode', +]); /** What the request says about a run, before the webhook node has produced anything. */ export type EngineV2WebhookRequest = { @@ -122,7 +126,6 @@ export class EngineV2Webhooks { ); } - // 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.`, @@ -132,7 +135,7 @@ export class EngineV2Webhooks { /** Converts the data plane's answer to the shape the v1 response path reads. */ async toRun( - outcome: Exclude, + outcome: Exclude, executionMode: WorkflowExecuteMode, ): Promise { const runData: IRunData = {}; diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index effeb40bffd..a7736b9a5c9 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -4,6 +4,7 @@ import type { Project } from '@n8n/db'; import { UserRepository } from '@n8n/db'; import { Container } from '@n8n/di'; import { ensureError } from '@n8n/utils/errors/ensure-error'; +import { isRecord } from '@n8n/utils/is-record'; import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import type express from 'express'; import merge from 'lodash/merge'; @@ -417,6 +418,76 @@ export const handleFormRedirectionCase = ( const { formDataFileSizeMax } = Container.get(GlobalConfig).endpoints; const parseFormData = createMultiFormDataParser(formDataFileSizeMax); +function isHttpFullResponse(response: unknown): response is IN8nHttpFullResponse { + // A missing body means that the response has an empty body. + return ( + isRecord(response) && isRecord(response.headers) && typeof response.statusCode === 'number' + ); +} + +async function sendResponseNodeResponse( + response: IN8nHttpFullResponse, + res: express.Response, + responseCallback: (error: Error | null, data: IWebhookResponseCallbackData) => void, + workflowStartNode: INode, + executionId: string | undefined, + workflow: Workflow, +): Promise { + try { + const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData; + if (binaryData?.id) { + if (response.statusCode) { + res.status(response.statusCode); + } + WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res); + applySandboxCSP(res); + try { + const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id); + res.once('close', () => stream.destroy()); + stream.pipe(res, { end: false }); + await finished(stream); + } finally { + void Container.get(WebhookResponseRelay).deleteOffloadedBody(response, { + workflowId: workflow.id, + executionId, + }); + } + responseCallback(null, { noWebhookResponse: true }); + } else if (Buffer.isBuffer(response.body)) { + if (response.statusCode) { + res.status(response.statusCode); + } + WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res); + applySandboxCSP(res); + res.end(response.body); + responseCallback(null, { noWebhookResponse: true }); + } else { + // TODO: This probably needs some more changes depending on the options on the + // Webhook Response node + + let data: IWebhookResponseCallbackData = { + data: response.body as IDataObject, + headers: response.headers, + responseCode: response.statusCode, + }; + + data = handleFormRedirectionCase(data, workflowStartNode); + + responseCallback(null, data); + } + + process.nextTick(() => res.end()); + } catch (e: unknown) { + const error = ensureError(e); + Container.get(ErrorReporter).error(error); + Container.get(Logger).error( + `Error with Webhook-Response for execution "${executionId}": "${error.message}"`, + { executionId, workflowId: workflow.id }, + ); + responseCallback(error, {}); + } +} + export function setupResponseNodePromise( responsePromise: IDeferredPromise, res: express.Response, @@ -426,7 +497,7 @@ export function setupResponseNodePromise( workflow: Workflow, ): void { void responsePromise.promise - .then(async (response: IN8nHttpFullResponse) => { + .then(async (response) => { if (response === EXECUTION_ENDED_WITHOUT_RESPONSE) { // The execution ended without the Respond to Webhook node running. The // post-execute handler answers instead, because only it knows whether the @@ -434,49 +505,14 @@ export function setupResponseNodePromise( return; } - const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData; - if (binaryData?.id) { - if (response.statusCode) { - res.status(response.statusCode); - } - WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res); - applySandboxCSP(res); - try { - const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id); - res.once('close', () => stream.destroy()); - stream.pipe(res, { end: false }); - await finished(stream); - } finally { - void Container.get(WebhookResponseRelay).deleteOffloadedBody(response, { - workflowId: workflow.id, - executionId, - }); - } - responseCallback(null, { noWebhookResponse: true }); - } else if (Buffer.isBuffer(response.body)) { - if (response.statusCode) { - res.status(response.statusCode); - } - WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res); - applySandboxCSP(res); - res.end(response.body); - responseCallback(null, { noWebhookResponse: true }); - } else { - // TODO: This probably needs some more changes depending on the options on the - // Webhook Response node - - let data: IWebhookResponseCallbackData = { - data: response.body as IDataObject, - headers: response.headers, - responseCode: response.statusCode, - }; - - data = handleFormRedirectionCase(data, workflowStartNode); - - responseCallback(null, data); - } - - process.nextTick(() => res.end()); + await sendResponseNodeResponse( + response, + res, + responseCallback, + workflowStartNode, + executionId, + workflow, + ); }) .catch(async (e: unknown) => { const error = ensureError(e); @@ -1076,7 +1112,7 @@ export async function executeWebhook( if (didPublishMcpRelay) return undefined; let responsePromise: IDeferredPromise | undefined; - if (responseMode === 'responseNode') { + if (responseMode === 'responseNode' && !routesToEngineV2) { responsePromise = createDeferredPromise(); // Mark the request as answered as soon as the node produces a response, before // `setupResponseNodePromise` starts writing it. Streaming offloaded binary data @@ -1121,8 +1157,10 @@ export async function executeWebhook( // the run and the listener agree on it. if (routesToEngineV2 && responseMode !== 'onReceived') { const engineExecutionId = createExecutionIdV2(); - pendingEngineV2Response = - Container.get(EngineV2WebhookResponder).waitForResponse(engineExecutionId); + pendingEngineV2Response = Container.get(EngineV2WebhookResponder).waitForResponse( + engineExecutionId, + responseMode === 'responseNode', + ); runData.engineExecutionId = engineExecutionId; } @@ -1229,7 +1267,30 @@ export async function executeWebhook( const waitForDataPlaneRun = async (waiting: PendingWebhookResponse) => { try { const outcome = await waiting.settled; + if (outcome.status === 'response') { + if (!isHttpFullResponse(outcome.response)) { + throw new UnexpectedError('Engine 2.0 produced an invalid webhook response'); + } + + didSendResponse = true; + await sendResponseNodeResponse( + outcome.response, + res, + responseCallback, + workflowStartNode, + executionId, + workflow, + ); + return undefined; + } + if (outcome.status === 'completed' || outcome.status === 'failed') { + if (outcome.status === 'completed' && responseMode === 'responseNode') { + responseCallback(null, { data: undefined, responseCode }); + didSendResponse = true; + return undefined; + } + return await engineV2Webhooks.toRun(outcome, executionMode); } @@ -1325,7 +1386,7 @@ export async function executeWebhook( return runData; } - // in `responseNode` mode `responseCallback` is called by `responsePromise` + // In v1 `responseNode` mode, `responsePromise` calls `responseCallback`. if (responseMode === 'responseNode' && responsePromise) { await Promise.allSettled([responsePromise.promise]); if (!didSendResponse) { diff --git a/packages/cli/test/integration/engine-v2-webhooks.test.ts b/packages/cli/test/integration/engine-v2-webhooks.test.ts index b9abb1b85a5..dde299b7c36 100644 --- a/packages/cli/test/integration/engine-v2-webhooks.test.ts +++ b/packages/cli/test/integration/engine-v2-webhooks.test.ts @@ -145,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 = 'responseNode'; + trigger.parameters.responseMode = 'streaming'; const workflow = await createV2Workflow(trigger); await startListening(workflow.id); @@ -155,9 +155,7 @@ describe('webhook runs on engine 2.0', () => { .send({ order: 42 }); expect(response.statusCode).toBe(400); - expect(response.body.message).toContain( - "does not support the 'responseNode' response mode yet", - ); + expect(response.body.message).toContain("does not support the 'streaming' response mode yet"); expect(startExecution).not.toHaveBeenCalled(); }); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 56daeb49215..f02762bbc1e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2539,6 +2539,9 @@ importers: '@n8n/typeorm': specifier: workspace:* version: link:../typeorm + '@n8n/utils': + specifier: workspace:* + version: link:../utils express: specifier: 'catalog:' version: 5.1.0(supports-color@8.1.1)