feat(engine): Report execution response failures (#39001)

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';

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',

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),

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<typeof executionResponseSchema>;
 
+/** The intended response could not be produced or carried. */
+export type UndeliverableMessage = Extract<ExecutionResponse, { type: 'undeliverable' }>;
+
 /**
  * The run is over. Always the last response an execution sends.
  *

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';

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));

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;

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<void> {
 		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);
+	}
 }

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);

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', {

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 = {

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', () => {

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<WebhookRunOutcome, { status: 'timeout' }>,
+		outcome: Exclude<WebhookRunOutcome, { status: 'timeout' | 'undeliverable' }>,
 		executionMode: WorkflowExecuteMode,
 	): Promise<IRun> {
 		const runData: IRunData = {};

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)