From 972db68d6bd059d3900064eb144cd79919752152 Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Wed, 23 Sep 2026 16:54:39 +0000 Subject: [PATCH] feat(engine): Report execution response failures (#39001) --- packages/@n8n/engine/src/index.ts | 7 ++- .../execution-response.schema.test.ts | 32 +++++++++++ .../execution-response.schema.ts | 8 +++ .../execution-response.types.ts | 3 ++ .../@n8n/engine/src/response-channel/index.ts | 6 ++- .../in-memory-execution-response.test.ts | 54 ++++++++++++++++--- .../src/modules/engine-v2/engine-v2.module.ts | 8 ++- .../in-memory-execution-response-sender.ts | 51 ++++++++++++++++-- ...ngine-v2-webhook-responder.service.test.ts | 15 ++++++ .../engine-v2-webhook-responder.service.ts | 8 +++ .../src/services/pending-webhook-response.ts | 2 + .../__tests__/webhook-helpers.test.ts | 21 ++++++++ .../cli/src/webhooks/engine-v2-webhooks.ts | 2 +- packages/cli/src/webhooks/webhook-helpers.ts | 34 ++++++++---- 14 files changed, 224 insertions(+), 27 deletions(-) diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index 482d7843151..c1445d83dda 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/packages/@n8n/engine/src/index.ts @@ -38,7 +38,12 @@ export { executionResponseSchema, noopExecutionResponseSender, } from './response-channel'; -export type { EndedMessage, ExecutionResponse, ExecutionResponseSender } from './response-channel'; +export type { + EndedMessage, + ExecutionResponse, + ExecutionResponseSender, + UndeliverableMessage, +} from './response-channel'; export { UnimplementedError } from './common'; export type { JsonObject, JsonValue } from './common'; diff --git a/packages/@n8n/engine/src/response-channel/__tests__/execution-response.schema.test.ts b/packages/@n8n/engine/src/response-channel/__tests__/execution-response.schema.test.ts index 9ec6f8ec06e..80841f79700 100644 --- a/packages/@n8n/engine/src/response-channel/__tests__/execution-response.schema.test.ts +++ b/packages/@n8n/engine/src/response-channel/__tests__/execution-response.schema.test.ts @@ -19,6 +19,37 @@ describe('executionResponseSchema', () => { }); }); + it('accepts an undeliverable response', () => { + expect( + executionResponseSchema.parse({ + type: 'undeliverable', + executionId: 'exec-1', + error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }), + ).toEqual({ + type: 'undeliverable', + executionId: 'exec-1', + error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + }); + + it.each([ + [ + 'execution id', + { type: 'undeliverable', executionId: '', error: { code: 'CODE', message: 'Bad' } }, + ], + [ + 'error code', + { type: 'undeliverable', executionId: 'exec-1', error: { code: '', message: 'Bad' } }, + ], + [ + 'error message', + { type: 'undeliverable', executionId: 'exec-1', error: { code: 'CODE', message: '' } }, + ], + ])('rejects an undeliverable response without an %s', (_field, response) => { + expect(executionResponseSchema.safeParse(response).success).toBe(false); + }); + it('accepts a step that produced nothing', () => { const parsed = executionResponseSchema.parse( ended({ @@ -32,6 +63,7 @@ describe('executionResponseSchema', () => { }, }), ); + if (parsed.type !== 'ended') throw new Error('Expected an ended response'); expect(parsed.lastStep).toEqual({ nodeId: 'a', 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 567d3b02d7f..d4ca5901fef 100644 --- a/packages/@n8n/engine/src/response-channel/execution-response.schema.ts +++ b/packages/@n8n/engine/src/response-channel/execution-response.schema.ts @@ -10,6 +10,14 @@ import { SETTLED_STEP_STATUSES } from '../execution/execution.types'; * boundary, so it is parsed against this before a handler sees it. */ export const executionResponseSchema = z.discriminatedUnion('type', [ + z.object({ + type: z.literal('undeliverable'), + executionId: z.string().min(1), + error: z.object({ + code: z.string().min(1), + message: z.string().min(1), + }), + }), 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 799aeb0d9b0..859179deeb5 100644 --- a/packages/@n8n/engine/src/response-channel/execution-response.types.ts +++ b/packages/@n8n/engine/src/response-channel/execution-response.types.ts @@ -17,6 +17,9 @@ import type { executionResponseSchema } from './execution-response.schema'; */ export type ExecutionResponse = z.infer; +/** The intended response could not be produced or carried. */ +export type UndeliverableMessage = Extract; + /** * The run is over. Always the last response an execution sends. * diff --git a/packages/@n8n/engine/src/response-channel/index.ts b/packages/@n8n/engine/src/response-channel/index.ts index 25075612b04..6538e1d62ab 100644 --- a/packages/@n8n/engine/src/response-channel/index.ts +++ b/packages/@n8n/engine/src/response-channel/index.ts @@ -1,4 +1,8 @@ export { noopExecutionResponseSender } from './execution-response-sender'; export type { ExecutionResponseSender } from './execution-response-sender'; export { executionResponseSchema } from './execution-response.schema'; -export type { EndedMessage, ExecutionResponse } from './execution-response.types'; +export type { + EndedMessage, + ExecutionResponse, + UndeliverableMessage, +} from './execution-response.types'; 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 8a3804554c3..a5627d4fe26 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 @@ -17,7 +17,7 @@ describe('in-memory execution responses', () => { it('serializes a response before publishing it', () => { const channel = new InMemoryExecutionResponseChannel(); const publish = vi.spyOn(channel, 'publish'); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); sender.send(ended('exec-1', [[{ at: new Date(0) }]])); @@ -27,6 +27,48 @@ describe('in-memory execution responses', () => { ); }); + it('reports a response that exceeds the frame size limit', () => { + const channel = new InMemoryExecutionResponseChannel(); + const publish = vi.spyOn(channel, 'publish'); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger(), 256); + + sender.send(ended('exec-1', [['x'.repeat(500)]])); + + expect(publish).toHaveBeenCalledExactlyOnceWith( + 'exec-1', + JSON.stringify({ + type: 'undeliverable', + executionId: 'exec-1', + error: { + code: 'RESPONSE_TOO_LARGE', + message: 'The execution response exceeds the maximum size of 256 bytes.', + }, + }), + ); + }); + + it('reports a response that cannot be serialized', () => { + const channel = new InMemoryExecutionResponseChannel(); + const publish = vi.spyOn(channel, 'publish'); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); + const response = ended() as ExecutionResponse & { circular?: unknown }; + response.circular = response; + + sender.send(response); + + expect(publish).toHaveBeenCalledExactlyOnceWith( + 'exec-1', + JSON.stringify({ + type: 'undeliverable', + executionId: 'exec-1', + error: { + code: 'RESPONSE_SERIALIZATION_FAILED', + message: 'The execution response could not be serialized.', + }, + }), + ); + }); + it('validates a response before delivering it', () => { const channel = new InMemoryExecutionResponseChannel(); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); @@ -62,7 +104,7 @@ describe('in-memory execution responses', () => { it('delivers a response to every handler for that execution', () => { const channel = new InMemoryExecutionResponseChannel(); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); const first: ExecutionResponse[] = []; const second: ExecutionResponse[] = []; @@ -77,7 +119,7 @@ describe('in-memory execution responses', () => { it('does not deliver a response to another execution', () => { const channel = new InMemoryExecutionResponseChannel(); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); const seen: ExecutionResponse[] = []; receiver.receive('exec-1', (response) => seen.push(response)); @@ -89,7 +131,7 @@ describe('in-memory execution responses', () => { it('stops delivery after the handler is removed', () => { const channel = new InMemoryExecutionResponseChannel(); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); const seen: ExecutionResponse[] = []; const unsubscribe = receiver.receive('exec-1', (response) => seen.push(response)); @@ -102,7 +144,7 @@ describe('in-memory execution responses', () => { it('stops publishing through the sender', async () => { const channel = new InMemoryExecutionResponseChannel(); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); const seen: string[] = []; channel.subscribe('exec-1', (frame) => seen.push(frame)); @@ -114,7 +156,7 @@ describe('in-memory execution responses', () => { it('stops delivery through the receiver', async () => { const channel = new InMemoryExecutionResponseChannel(); - const sender = new InMemoryExecutionResponseSender(channel); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); const seen: ExecutionResponse[] = []; receiver.receive('exec-1', (response) => seen.push(response)); 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 3cc8321b65b..d491b0ec046 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.module.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.module.ts @@ -56,11 +56,9 @@ export class EngineV2Module implements ModuleInterface { // 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(); - const responseSender = new InMemoryExecutionResponseSender(responseChannel); - const responseReceiver = new InMemoryExecutionResponseReceiver( - responseChannel, - Container.get(Logger).scoped('engine-v2'), - ); + const scopedLogger = Container.get(Logger).scoped('engine-v2'); + const responseSender = new InMemoryExecutionResponseSender(responseChannel, scopedLogger); + const responseReceiver = new InMemoryExecutionResponseReceiver(responseChannel, scopedLogger); Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver); this.responseSender = responseSender; this.responseReceiver = responseReceiver; 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 0645e72c7ce..75677236e09 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,19 +1,64 @@ -import type { ExecutionResponse, ExecutionResponseSender } from '@n8n/engine'; +import type { Logger } from '@n8n/backend-common'; +import type { ExecutionResponse, ExecutionResponseSender, UndeliverableMessage } from '@n8n/engine'; +import { toResult } from '@n8n/utils/result'; 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; + export class InMemoryExecutionResponseSender implements ExecutionResponseSender { private stopped = false; - constructor(private readonly channel: InMemoryExecutionResponseChannel) {} + constructor( + private readonly channel: InMemoryExecutionResponseChannel, + private readonly logger: Logger, + private readonly maxFrameBytes: number = MAX_FRAME_BYTES, + ) {} send(response: ExecutionResponse): void { if (this.stopped) return; - this.channel.publish(response.executionId, JSON.stringify(response)); + this.channel.publish(response.executionId, this.toFrame(response)); } async stop(): Promise { this.stopped = true; } + + private toFrame(response: ExecutionResponse): string { + const serialized = toResult(() => JSON.stringify(response)); + if (!serialized.ok) { + this.logger.warn('Could not serialize an execution response', { + executionId: response.executionId, + type: response.type, + error: serialized.error, + }); + return this.undeliverableFrame(response.executionId, { + code: 'RESPONSE_SERIALIZATION_FAILED', + message: 'The execution response could not be serialized.', + }); + } + + if (Buffer.byteLength(serialized.result) > this.maxFrameBytes) { + this.logger.warn('Execution response exceeds the frame size limit', { + executionId: response.executionId, + type: response.type, + }); + return this.undeliverableFrame(response.executionId, { + code: 'RESPONSE_TOO_LARGE', + message: `The execution response exceeds the maximum size of ${this.maxFrameBytes} bytes.`, + }); + } + + return serialized.result; + } + + private undeliverableFrame(executionId: string, error: UndeliverableMessage['error']): string { + return JSON.stringify({ + type: 'undeliverable', + executionId, + error, + } satisfies UndeliverableMessage); + } } 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 e8dc9771528..a39f2387b26 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 @@ -159,6 +159,21 @@ describe('EngineV2WebhookResponder', () => { }); }); + it('reports a response failure without attributing it to a node', async () => { + const pending = responder.waitForResponse(createExecutionIdV2()); + + deliver({ + type: 'undeliverable', + executionId: pending.executionId, + error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + + await expect(pending.settled).resolves.toEqual({ + status: 'undeliverable', + error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + }); + it('times out rather than waiting forever for a lost answer', async () => { const impatient = newResponder(1); impatient.useReceiver(fakeReceiver().receiver); 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 6cacec910df..abe12dda07b 100644 --- a/packages/cli/src/services/engine-v2-webhook-responder.service.ts +++ b/packages/cli/src/services/engine-v2-webhook-responder.service.ts @@ -87,6 +87,14 @@ 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); } catch (error) { this.logger.error('Failed to relay an engine 2.0 response', { diff --git a/packages/cli/src/services/pending-webhook-response.ts b/packages/cli/src/services/pending-webhook-response.ts index 8df5692297d..f5e6dc69a9e 100644 --- a/packages/cli/src/services/pending-webhook-response.ts +++ b/packages/cli/src/services/pending-webhook-response.ts @@ -10,6 +10,8 @@ export type LastNode = { nodeName: string; outputs: StepSlots }; export type WebhookRunOutcome = | { status: 'completed'; lastNode?: LastNode } | { status: 'failed'; nodeName: string; error?: { name: string; message: string } } + /** The execution response could not be produced or delivered. */ + | { status: 'undeliverable'; error: { name: string; message: string } } | { status: 'timeout' }; export type PendingWebhookResponseOptions = { diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index a9440d4e27f..dd98d06d17d 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts @@ -2224,6 +2224,27 @@ describe('executeWebhook on engine 2.0', () => { }, ]); }); + + it('answers with the channel error when the response is undeliverable', 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: 'undeliverable', + error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + + await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1)); + expect(responseCallback.mock.calls[0]).toEqual([ + null, + { + data: { message: 'The response is too large.' }, + responseCode: 500, + }, + ]); + }); }); describe('rejections', () => { diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index 148940e437c..e340204123c 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -132,7 +132,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 d4b1d00719b..2b8db9bf1de 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -831,7 +831,7 @@ export async function executeWebhook( }; let didSendResponse = false; - /** Whether this run goes to the engine 2.0 data plane instead of the v1 path. */ + /** Whether this run goes to the engine v2 data plane instead of the v1 path. */ let routesToEngineV2 = false; let pendingEngineV2Response: PendingWebhookResponse | undefined; let runExecutionDataMerge = {}; @@ -1052,7 +1052,7 @@ export async function executeWebhook( userId: webhookData.userId, encryptedRunnerIdentity: additionalData.encryptedRunnerIdentity, // v1 reads this from `executionData.startData`, which `prepareExecutionData` - // sets, so carrying it here changes nothing for v1. Engine 2.0 has no way to + // sets, so carrying it here changes nothing for v1. Engine v2 has no way to // stop at a node, and its dispatcher refuses the run on this field. destinationNode, }; @@ -1209,26 +1209,40 @@ export async function executeWebhook( /** * 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. + * Returns the run data when it arrives. Otherwise, it sends the response + * channel error or a 504 timeout to the webhook caller. */ const waitForDataPlaneRun = async (waiting: PendingWebhookResponse) => { try { const outcome = await waiting.settled; - if (outcome.status !== 'timeout') { + if (outcome.status === 'completed' || outcome.status === 'failed') { return await engineV2Webhooks.toRun(outcome, executionMode); } - Container.get(Logger).warn('No answer arrived for an engine 2.0 webhook run', { + const isUndeliverable = outcome.status === 'undeliverable'; + const errorResponse = isUndeliverable + ? { + logMessage: 'Could not deliver an engine v2 webhook response', + responseMessage: outcome.error.message, + responseCode: 500, + } + : { + // timeout + logMessage: 'No answer arrived for an engine v2 webhook run', + responseMessage: 'The workflow did not answer in time', + responseCode: 504, + }; + Container.get(Logger).warn(errorResponse.logMessage, { executionId, workflowId: workflowData.id, + ...(isUndeliverable ? { error: outcome.error } : {}), }); // The webhook node can answer before the execution starts. Do not send a - // second response if the execution response later times out. + // second response when the execution response later settles. if (!didSendResponse) { responseCallback(null, { - data: { message: 'The workflow did not answer in time' }, - responseCode: 504, + data: { message: errorResponse.responseMessage }, + responseCode: errorResponse.responseCode, }); didSendResponse = true; } @@ -1240,7 +1254,7 @@ export async function executeWebhook( }; // 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 + // Engine v2 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)