refactor(engine): Split the engine v2 webhook responder into a registry and an outcome mapping (no-changelog) (#39978)

packages/cli/src/modules/engine-v2/__tests__/buffer-webhook-response.test.ts

@@ -11,14 +11,14 @@ import type { IWorkflowExecuteAdditionalData } from 'n8n-workflow';
 import { mock } from 'vitest-mock-extended';
 
 import { createExecutionIdV2 } from '@/executions/execution-id';
-import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
+import { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 
 import { InMemoryExecutionResponseChannel } from '../response-channel/in-memory-execution-response-channel';
 import { InMemoryExecutionResponseReceiver } from '../response-channel/in-memory-execution-response-receiver';
 import { InMemoryExecutionResponseSender } from '../response-channel/in-memory-execution-response-sender';
 
 /**
- * Runs `sendResponse` through the real hook, sender, receiver and responder.
+ * Runs `sendResponse` through the real hook, sender, receiver and registry.
  * The unit tests for each of them check one step. This checks that the Buffer
  * body that goes in is the one that comes out.
  */
@@ -31,14 +31,14 @@ describe('a Buffer webhook response through the response channel', () => {
 		const sender = new InMemoryExecutionResponseSender(channel, mockLogger());
 		const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
 
-		const responder = new EngineV2WebhookResponder(
+		const registry = new EngineV2WebhookResponseRegistry(
 			mock<EngineConfig>({ webhookResponseTimeout: 50_000 }),
 			mock<Logger>({ scoped: () => mock<Logger>() }),
 		);
-		responder.useReceiver(receiver);
+		registry.useReceiver(receiver);
 
 		const executionId = createExecutionIdV2();
-		const pending = await responder.waitForResponse(executionId, expectation);
+		const pending = await registry.waitForResponse(executionId, expectation);
 
 		const additionalData = {} as IWorkflowExecuteAdditionalData;
 		attachResponseHooks(additionalData, {
@@ -64,7 +64,7 @@ describe('a Buffer webhook response through the response channel', () => {
 			{ body: Buffer.from(bytes), headers, statusCode: 201 },
 		]);
 
-		const outcome = await pending.settled;
+		const outcome = await pending.outcome;
 		expect(outcome).toEqual({
 			status: 'response',
 			response: { body: bytes, headers, statusCode: 201 },
@@ -80,7 +80,7 @@ describe('a Buffer webhook response through the response channel', () => {
 			{ body: { ok: true }, headers: {}, statusCode: 200 },
 		]);
 
-		await expect(pending.settled).resolves.toEqual({
+		await expect(pending.outcome).resolves.toEqual({
 			status: 'response',
 			response: { body: { ok: true }, headers: {}, statusCode: 200 },
 		});

packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts

@@ -4,7 +4,7 @@ import { Container } from '@n8n/di';
 import { mock } from 'vitest-mock-extended';
 
 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
-import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
+import { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 import { RedisClientService } from '@n8n/backend-services';
 
 import { EngineControlPlaneServer } from '../engine-control-plane-server';
@@ -22,7 +22,7 @@ describe('EngineV2Module', () => {
 	let runtime: EngineV2Runtime;
 	let client: EngineDataPlaneClient;
 	let controlPlaneServer: EngineControlPlaneServer;
-	let webhookResponder: EngineV2WebhookResponder;
+	let webhookResponseRegistry: EngineV2WebhookResponseRegistry;
 	let redisClientService: RedisClientService;
 	let subscriber: ReturnType<typeof mock<RedisResponseSubscriber>>;
 
@@ -34,7 +34,7 @@ describe('EngineV2Module', () => {
 		runtime = mockInstance(EngineV2Runtime);
 		client = mockInstance(EngineDataPlaneClient);
 		controlPlaneServer = mockInstance(EngineControlPlaneServer);
-		webhookResponder = mockInstance(EngineV2WebhookResponder);
+		webhookResponseRegistry = mockInstance(EngineV2WebhookResponseRegistry);
 		subscriber = mock<RedisResponseSubscriber>();
 		mockInstance(GlobalConfig, { redis: mock<GlobalConfig['redis']>({ prefix: 'n8n' }) });
 		redisClientService = mockInstance(RedisClientService, {
@@ -89,7 +89,7 @@ describe('EngineV2Module', () => {
 		it('receives responses in memory and opens no Redis client', async () => {
 			await module.init();
 
-			expect(webhookResponder.useReceiver).toHaveBeenCalledWith(
+			expect(webhookResponseRegistry.useReceiver).toHaveBeenCalledWith(
 				expect.any(InMemoryExecutionResponseReceiver),
 			);
 			expect(runtime.init).toHaveBeenCalledWith(expect.anything());
@@ -146,7 +146,7 @@ describe('EngineV2Module', () => {
 			expect(redisClientService.createClient).toHaveBeenCalledTimes(1);
 			expect(redisClientService.createClient).toHaveBeenCalledWith({ type: 'subscriber(n8n)' });
 			expect(subscriber.on).toHaveBeenCalledWith('message', expect.any(Function));
-			expect(webhookResponder.useReceiver).toHaveBeenCalledWith(
+			expect(webhookResponseRegistry.useReceiver).toHaveBeenCalledWith(
 				expect.any(RedisExecutionResponseReceiver),
 			);
 		});
@@ -159,7 +159,7 @@ describe('EngineV2Module', () => {
 			await expect(module.init()).rejects.toThrow('Subscriber failed');
 
 			expect(subscriber.disconnect).toHaveBeenCalledTimes(1);
-			expect(webhookResponder.useReceiver).not.toHaveBeenCalled();
+			expect(webhookResponseRegistry.useReceiver).not.toHaveBeenCalled();
 		});
 
 		it('refuses to start without a shared secret', async () => {

packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts

@@ -0,0 +1,131 @@
+import type { ExecutionResponse, JsonValue } from '@n8n/engine';
+import { ENCODED_BUFFER_KEY } from 'n8n-core';
+
+import { toWebhookOutcome } from '@/modules/engine-v2/webhook-response/webhook-outcome';
+
+const runEnd = { kind: 'runEnd' } as const;
+const stepResponse = { kind: 'stepResponse' } as const;
+
+const ended = (overrides: Record<string, unknown> = {}): ExecutionResponse => ({
+	type: 'ended',
+	executionId: 'exec-1',
+	workflowId: 'wf-1',
+	status: 'completed',
+	lastStep: {
+		nodeId: 'a',
+		nodeName: 'A',
+		status: 'completed',
+		outputs: [[{ json: { a: 1 } }]],
+	},
+	...overrides,
+});
+
+const response = (payload: JsonValue): ExecutionResponse => ({
+	type: 'response',
+	executionId: 'exec-1',
+	payload,
+});
+
+describe('toWebhookOutcome', () => {
+	it('reports the step the run ended with', () => {
+		expect(toWebhookOutcome(ended(), runEnd)).toEqual({
+			status: 'completed',
+			lastNode: { nodeName: 'A', outputs: [[{ json: { a: 1 } }]] },
+		});
+	});
+
+	it('reports no last node when that step produced nothing', () => {
+		const outcome = toWebhookOutcome(
+			ended({ lastStep: { nodeId: 'c', nodeName: 'C', status: 'skipped', outputs: null } }),
+			runEnd,
+		);
+
+		expect(outcome).toEqual({ status: 'completed', lastNode: undefined });
+	});
+
+	it('reports a failure with the node that caused it', () => {
+		const outcome = toWebhookOutcome(
+			ended({
+				status: 'failed',
+				lastStep: {
+					nodeId: 'c',
+					nodeName: 'C',
+					status: 'failed',
+					outputs: null,
+					error: { name: 'NodeOperationError', message: 'it broke' },
+				},
+			}),
+			runEnd,
+		);
+
+		expect(outcome).toEqual({
+			status: 'failed',
+			nodeName: 'C',
+			error: { name: 'NodeOperationError', message: 'it broke' },
+		});
+	});
+
+	it('reports a response failure without attributing it to a node', () => {
+		const outcome = toWebhookOutcome(
+			{
+				type: 'undeliverable',
+				executionId: 'exec-1',
+				error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+			},
+			runEnd,
+		);
+
+		expect(outcome).toEqual({
+			status: 'undeliverable',
+			error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+		});
+	});
+
+	it('reports the response produced by the Respond node', () => {
+		const payload = { body: { ok: true }, headers: {}, statusCode: 200 };
+
+		expect(toWebhookOutcome(response(payload), stepResponse)).toEqual({
+			status: 'response',
+			response: payload,
+		});
+	});
+
+	it('restores a Buffer body the data plane sent as a base64 envelope', () => {
+		const bytes = Buffer.from([0x00, 0xff, 0x10]);
+		const headers = { 'content-type': 'application/octet-stream', 'content-length': 3 };
+
+		const outcome = toWebhookOutcome(
+			response({
+				body: { [ENCODED_BUFFER_KEY]: bytes.toString('base64') },
+				headers,
+				statusCode: 201,
+			}),
+			stepResponse,
+		);
+
+		expect(outcome).toEqual({
+			status: 'response',
+			response: { body: bytes, headers, statusCode: 201 },
+		});
+		expect(Buffer.isBuffer((outcome as { response: { body: unknown } }).response.body)).toBe(true);
+	});
+
+	it.each([runEnd, { kind: 'none' } as const])(
+		'ignores a Respond node result when the expectation is %j',
+		(expectation) => {
+			const payload = { body: { ignored: true }, headers: {}, statusCode: 200 };
+
+			expect(toWebhookOutcome(response(payload), expectation)).toBeUndefined();
+		},
+	);
+
+	it('does not settle on a chunk', () => {
+		const chunk: ExecutionResponse = {
+			type: 'chunk',
+			executionId: 'exec-1',
+			payload: { type: 'item', content: 'hi' },
+		};
+
+		expect(toWebhookOutcome(chunk, { kind: 'stream' })).toBeUndefined();
+	});
+});

packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts

@@ -2,16 +2,15 @@ import type { Logger } from '@n8n/backend-common';
 import type { EngineConfig } from '@n8n/config';
 import type { ExecutionResponse } from '@n8n/engine';
 import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
-import { ENCODED_BUFFER_KEY } from 'n8n-core';
 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,
+	EngineV2WebhookResponseRegistry,
 	MAX_PENDING_WEBHOOKS,
 	SUBSCRIBE_TIMEOUT_MS,
-} from '@/services/engine-v2-webhook-responder.service';
+} from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 
 const TIMEOUT_MS = 50_000;
 
@@ -21,7 +20,7 @@ const stepResponse = { kind: 'stepResponse' } as const;
 /**
  * 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.
+ * trip; what matters here is what the registry does with a response.
  *
  * Handlers are per execution, as the real receiver's are. A response for
  * another run reaches nobody.
@@ -58,141 +57,105 @@ const endedResponse = (executionId: string, overrides: Record<string, unknown> =
 	...overrides,
 });
 
-const newResponder = (timeoutMs = TIMEOUT_MS) =>
-	new EngineV2WebhookResponder(
+const newRegistry = (timeoutMs = TIMEOUT_MS) =>
+	new EngineV2WebhookResponseRegistry(
 		mock<EngineConfig>({ webhookResponseTimeout: timeoutMs }),
 		mock<Logger>({ scoped: () => mock<Logger>() }),
 	);
 
-describe('EngineV2WebhookResponder', () => {
+describe('EngineV2WebhookResponseRegistry', () => {
 	let deliver: (response: ExecutionResponse) => void;
-	let responder: EngineV2WebhookResponder;
+	let registry: EngineV2WebhookResponseRegistry;
 
 	beforeEach(() => {
 		vi.useFakeTimers();
 		const fake = fakeReceiver();
 		deliver = fake.deliver;
-		responder = newResponder();
-		responder.useReceiver(fake.receiver);
+		registry = newRegistry();
+		registry.useReceiver(fake.receiver);
 	});
 
 	afterEach(() => {
 		vi.useRealTimers();
 	});
 
+	const stillPending = Symbol('still pending');
+	const raceWithPending = async (outcome: Promise<unknown>) =>
+		await Promise.race([outcome, Promise.resolve(stillPending)]);
+
 	it('listens under the id the run is started with', async () => {
 		const executionId = createExecutionIdV2();
 
-		expect((await responder.waitForResponse(executionId, runEnd)).executionId).toBe(executionId);
+		expect((await registry.waitForResponse(executionId, runEnd)).executionId).toBe(executionId);
 	});
 
 	it('refuses to listen before the host hands over a receiver', async () => {
-		await expect(newResponder().waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
+		await expect(newRegistry().waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
 			'without a receiver',
 		);
 	});
 
 	it('refuses a run once it listens for as many as it can hold', async () => {
 		for (let i = 0; i < MAX_PENDING_WEBHOOKS; i++) {
-			await responder.waitForResponse(createExecutionIdV2(), runEnd);
+			await registry.waitForResponse(createExecutionIdV2(), runEnd);
 		}
 
 		// Refused before dispatch, so no run starts that nothing can answer.
-		await expect(responder.waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
+		await expect(registry.waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
 			'Try again later',
 		);
 	});
 
-	it('listens again once an answered run releases its slot', async () => {
-		const pending = await Promise.all(
+	const fillToCapacity = async () =>
+		await Promise.all(
 			Array.from(
 				{ length: MAX_PENDING_WEBHOOKS },
-				async () => await responder.waitForResponse(createExecutionIdV2(), runEnd),
+				async () => await registry.waitForResponse(createExecutionIdV2(), runEnd),
 			),
 		);
 
+	it('listens again once a released run frees its slot', async () => {
+		const pending = await fillToCapacity();
+
 		pending[0].release();
 
-		await expect(responder.waitForResponse(createExecutionIdV2(), runEnd)).resolves.toBeDefined();
+		await expect(registry.waitForResponse(createExecutionIdV2(), runEnd)).resolves.toBeDefined();
+	});
+
+	it('frees the slot and the timer once the run answers', async () => {
+		const pending = await fillToCapacity();
+
+		deliver(endedResponse(pending[0].executionId));
+		await pending[0].outcome;
+
+		await expect(registry.waitForResponse(createExecutionIdV2(), runEnd)).resolves.toBeDefined();
+		// Only the timers of the runs that still wait remain.
+		expect(vi.getTimerCount()).toBe(MAX_PENDING_WEBHOOKS);
 	});
 
 	it('leaves an unrelated pending run unaffected', async () => {
-		const other = await responder.waitForResponse(createExecutionIdV2(), runEnd);
-		const pending = await responder.waitForResponse(createExecutionIdV2(), runEnd);
+		const other = await registry.waitForResponse(createExecutionIdV2(), runEnd);
+		const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd);
 
 		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,
-		);
+		await expect(pending.outcome).resolves.toMatchObject({ status: 'completed' });
+		await expect(raceWithPending(other.outcome)).resolves.toBe(stillPending);
 	});
 
-	it('reports the step the run ended with', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), runEnd);
+	it('settles with the outcome of the response', async () => {
+		const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd);
 
 		deliver(endedResponse(pending.executionId));
 
-		await expect(pending.settled).resolves.toEqual({
+		await expect(pending.outcome).resolves.toEqual({
 			status: 'completed',
 			lastNode: { nodeName: 'A', outputs: [[{ json: { a: 1 } }]] },
 		});
 	});
 
-	it('reports no last node when that step produced nothing', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), runEnd);
-
-		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 the response produced by the Respond node', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), stepResponse);
-
-		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('restores a Buffer body the data plane sent as a base64 envelope', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), stepResponse);
-		const bytes = Buffer.from([0x00, 0xff, 0x10]);
-		const headers = { 'content-type': 'application/octet-stream', 'content-length': 3 };
-
-		deliver({
-			type: 'response',
-			executionId: pending.executionId,
-			payload: {
-				body: { [ENCODED_BUFFER_KEY]: bytes.toString('base64') },
-				headers,
-				statusCode: 201,
-			},
-		});
-
-		const outcome = await pending.settled;
-		expect(outcome).toEqual({
-			status: 'response',
-			response: { body: bytes, headers, statusCode: 201 },
-		});
-		expect(Buffer.isBuffer((outcome as { response: { body: unknown } }).response.body)).toBe(true);
-	});
-
 	it('keeps the first terminal outcome', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), stepResponse);
+		const pending = await registry.waitForResponse(createExecutionIdV2(), stepResponse);
 
 		deliver({
 			type: 'response',
@@ -201,96 +164,73 @@ describe('EngineV2WebhookResponder', () => {
 		});
 		deliver(endedResponse(pending.executionId));
 
-		await expect(pending.settled).resolves.toMatchObject({ status: 'response' });
+		await expect(pending.outcome).resolves.toMatchObject({ status: 'response' });
 	});
 
-	it.each([runEnd, { kind: 'none' } as const])(
-		'ignores a Respond node result when the expectation is %j',
-		async (expectation) => {
-			const pending = await responder.waitForResponse(createExecutionIdV2(), expectation);
-
-			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 = await responder.waitForResponse(createExecutionIdV2(), runEnd);
-
-		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('reports a response failure without attributing it to a node', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), runEnd);
+	it('keeps waiting when a response does not answer the expectation', async () => {
+		const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd);
 
 		deliver({
-			type: 'undeliverable',
+			type: 'response',
 			executionId: pending.executionId,
-			error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+			payload: { body: { ignored: true }, headers: {}, statusCode: 200 },
 		});
+		await expect(raceWithPending(pending.outcome)).resolves.toBe(stillPending);
 
-		await expect(pending.settled).resolves.toEqual({
-			status: 'undeliverable',
-			error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
-		});
+		deliver(endedResponse(pending.executionId));
+		await expect(pending.outcome).resolves.toMatchObject({ status: 'completed' });
 	});
 
 	it('times out rather than waiting forever for a lost answer', async () => {
-		const impatient = newResponder(1);
+		const impatient = newRegistry(1);
 		impatient.useReceiver(fakeReceiver().receiver);
 
 		const pending = await impatient.waitForResponse(createExecutionIdV2(), runEnd);
 		await vi.advanceTimersByTimeAsync(1);
-		await expect(pending.settled).resolves.toEqual({
+		await expect(pending.outcome).resolves.toEqual({
 			status: 'timeout',
 		});
+
+		// The timed-out run frees its slot on its own.
+		await expect(impatient.waitForResponse(pending.executionId, runEnd)).resolves.toBeDefined();
 	});
 
 	it('drops later responses for a released run', async () => {
-		const pending = await responder.waitForResponse(createExecutionIdV2(), runEnd);
+		const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd);
 		pending.release();
 
 		deliver(endedResponse(pending.executionId));
 
-		await expect(Promise.race([pending.settled, Promise.resolve('still waiting')])).resolves.toBe(
-			'still waiting',
+		await expect(raceWithPending(pending.outcome)).resolves.toBe(stillPending);
+	});
+
+	it('ignores a late release after the slot is taken again', async () => {
+		const executionId = createExecutionIdV2();
+		const first = await registry.waitForResponse(executionId, runEnd);
+		deliver(endedResponse(executionId));
+		await first.outcome;
+
+		const second = await registry.waitForResponse(executionId, runEnd);
+		first.release();
+
+		await expect(registry.waitForResponse(executionId, runEnd)).rejects.toThrow(
+			'already waits for a response for this execution',
 		);
+		deliver(endedResponse(executionId));
+		await expect(second.outcome).resolves.toMatchObject({ status: 'completed' });
 	});
 
 	it('refuses a second wait for the same execution', async () => {
 		const executionId = createExecutionIdV2();
-		await responder.waitForResponse(executionId, runEnd);
+		await registry.waitForResponse(executionId, runEnd);
 
-		await expect(responder.waitForResponse(executionId, runEnd)).rejects.toThrow(
+		await expect(registry.waitForResponse(executionId, runEnd)).rejects.toThrow(
 			'already waits for a response for this execution',
 		);
 	});
 });
 
-describe('EngineV2WebhookResponder while the receiver subscribes', () => {
+describe('EngineV2WebhookResponseRegistry while the receiver subscribes', () => {
 	/** A receiver whose subscriptions complete only when the test says so. */
 	function slowReceiver() {
 		const subscriptions: Array<IDeferredPromise<() => void>> = [];
@@ -318,46 +258,63 @@ describe('EngineV2WebhookResponder while the receiver subscribes', () => {
 
 	it('counts runs that are still subscribing against the limit', async () => {
 		const { receiver } = slowReceiver();
-		const responder = newResponder();
-		responder.useReceiver(receiver);
+		const registry = newRegistry();
+		registry.useReceiver(receiver);
 
 		for (let i = 0; i < MAX_PENDING_WEBHOOKS; i++) {
-			void responder.waitForResponse(createExecutionIdV2(), runEnd);
+			void registry.waitForResponse(createExecutionIdV2(), runEnd);
 		}
 
-		await expect(responder.waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
+		await expect(registry.waitForResponse(createExecutionIdV2(), runEnd)).rejects.toThrow(
 			'Try again later',
 		);
 	});
 
 	it('frees the slot and the timer when the subscription fails', async () => {
 		const { receiver, subscriptions } = slowReceiver();
-		const responder = newResponder();
-		responder.useReceiver(receiver);
+		const registry = newRegistry();
+		registry.useReceiver(receiver);
 		const executionId = createExecutionIdV2();
 
-		const failed = responder.waitForResponse(executionId, runEnd);
+		const failed = registry.waitForResponse(executionId, runEnd);
 		await vi.waitFor(() => expect(subscriptions).toHaveLength(1));
 		subscriptions[0].reject(new Error('Redis is unavailable'));
 
 		await expect(failed).rejects.toThrow('Redis is unavailable');
 		expect(vi.getTimerCount()).toBe(0);
 
-		const retried = responder.waitForResponse(executionId, runEnd);
+		const retried = registry.waitForResponse(executionId, runEnd);
 		await vi.waitFor(() => expect(subscriptions).toHaveLength(2));
 		subscriptions[1].resolve(() => {});
 		await expect(retried).resolves.toBeDefined();
 	});
 
+	it('drops the subscription when the wait settles before it is ready', async () => {
+		const { receiver, subscriptions } = slowReceiver();
+		const registry = newRegistry(1_000);
+		registry.useReceiver(receiver);
+
+		const waiting = registry.waitForResponse(createExecutionIdV2(), runEnd);
+		await vi.waitFor(() => expect(subscriptions).toHaveLength(1));
+		await vi.advanceTimersByTimeAsync(1_000);
+
+		const unsubscribe = vi.fn();
+		subscriptions[0].resolve(unsubscribe);
+
+		const pending = await waiting;
+		await expect(pending.outcome).resolves.toEqual({ status: 'timeout' });
+		expect(unsubscribe).toHaveBeenCalledTimes(1);
+	});
+
 	it('gives up at the subscribe timeout and drops a late subscription', async () => {
 		const { receiver, subscriptions } = slowReceiver();
 		// The response timeout is shorter, but it must not end the subscribe wait.
-		const responder = newResponder(1_000);
-		responder.useReceiver(receiver);
+		const registry = newRegistry(1_000);
+		registry.useReceiver(receiver);
 		const executionId = createExecutionIdV2();
 
 		let settled = false;
-		const waiting = responder.waitForResponse(executionId, runEnd).finally(() => {
+		const waiting = registry.waitForResponse(executionId, runEnd).finally(() => {
 			settled = true;
 		});
 		const rejection = expect(waiting).rejects.toThrow(`within ${SUBSCRIBE_TIMEOUT_MS / 1000}s`);
@@ -370,7 +327,7 @@ describe('EngineV2WebhookResponder while the receiver subscribes', () => {
 		expect(vi.getTimerCount()).toBe(0);
 
 		// The slot is free before the late subscription completes.
-		const retried = responder.waitForResponse(executionId, runEnd);
+		const retried = registry.waitForResponse(executionId, runEnd);
 		await vi.waitFor(() => expect(subscriptions).toHaveLength(2));
 
 		const unsubscribe = vi.fn();

packages/cli/src/modules/engine-v2/engine-v2.module.ts

@@ -47,15 +47,15 @@ export class EngineV2Module implements ModuleInterface {
 		await Container.get(EngineControlPlaneServer).start();
 
 		const logger = Container.get(Logger).scoped('engine-v2');
-		const { EngineV2WebhookResponder } = await import(
-			'@/services/engine-v2-webhook-responder.service.js'
+		const { EngineV2WebhookResponseRegistry } = await import(
+			'./webhook-response/webhook-response-registry.service.js'
 		);
 
 		if (engineConfig.mode === 'in-process') {
 			// Hand both endpoints over before the engine starts. A short run can answer
 			// before `startExecution` returns, and responses are not replayed.
 			const { responseSender, responseReceiver } = await this.initInMemoryResponseChannel(logger);
-			Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver);
+			Container.get(EngineV2WebhookResponseRegistry).useReceiver(responseReceiver);
 			this.responseSender = responseSender;
 			this.responseReceiver = responseReceiver;
 
@@ -68,7 +68,7 @@ export class EngineV2Module implements ModuleInterface {
 				'./response-channel/redis-execution-response-channel.js'
 			);
 			const responseReceiver = await startRedisExecutionResponseReceiver(logger);
-			Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver);
+			Container.get(EngineV2WebhookResponseRegistry).useReceiver(responseReceiver);
 			this.responseReceiver = responseReceiver;
 		}
 

packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts

@@ -0,0 +1,73 @@
+import type { ExecutionResponse, ResponseExpectation, StepSlots } from '@n8n/engine';
+import { decodeBufferBody } from 'n8n-core';
+import { UnexpectedError } from 'n8n-workflow';
+
+/** 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: '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. */
+	| { status: 'undeliverable'; error: { name: string; message: string } }
+	| { status: 'timeout' };
+
+/**
+ * Maps one execution response to the outcome of the webhook request that waits
+ * for it. Returns `undefined` when the response does not answer the request.
+ *
+ * @param expectation What the request waits for. Only `stepResponse` lets a
+ * Respond to Webhook result answer it. The engine obeys this too, so this
+ * filter is a second defense.
+ */
+export function toWebhookOutcome(
+	received: ExecutionResponse,
+	expectation: ResponseExpectation,
+): WebhookRunOutcome | undefined {
+	switch (received.type) {
+		case 'undeliverable':
+			return {
+				status: 'undeliverable',
+				error: { name: received.error.code, message: received.error.message },
+			};
+
+		case 'response':
+			if (expectation.kind !== 'stepResponse') return undefined;
+			// A Buffer body arrives base64-encoded, because the channel is JSON.
+			return { status: 'response', response: decodeBufferBody(received.payload) };
+
+		case 'chunk':
+			// A chunk is a part of a stream. It never answers the request alone.
+			return undefined;
+
+		case 'ended': {
+			const { nodeName, outputs, error } = received.lastStep;
+
+			switch (received.status) {
+				case 'failed':
+					// The step that ended a failed run is the one that failed, so its name
+					// and error are what the caller reports.
+					return { status: 'failed', nodeName, error };
+
+				case 'completed':
+					return {
+						status: 'completed',
+						// A skipped or failed step carries nothing to answer with.
+						lastNode: outputs ? { nodeName, outputs } : undefined,
+					};
+
+				default: {
+					const exhaustive: never = received.status;
+					throw new UnexpectedError(`Unexpected run status: ${JSON.stringify(exhaustive)}`);
+				}
+			}
+		}
+
+		default: {
+			const exhaustive: never = received;
+			throw new UnexpectedError(`Unexpected response type: ${JSON.stringify(exhaustive)}`);
+		}
+	}
+}

packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts

@@ -1,21 +1,40 @@
 import { Logger } from '@n8n/backend-common';
 import { EngineConfig } from '@n8n/config';
 import { Service } from '@n8n/di';
-import type { EndedMessage, ExecutionResponse, ResponseExpectation } from '@n8n/engine';
-import { decodeBufferBody } from 'n8n-core';
+import type { ExecutionResponse, ResponseExpectation } from '@n8n/engine';
+import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
 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';
+} from '../response-channel/execution-response-receiver';
+import { toWebhookOutcome, type WebhookRunOutcome } from './webhook-outcome';
+
+/** The caller's view of an open webhook request that waits for its run. */
+export interface WebhookResponseWait {
+	readonly executionId: ExecutionIdV2;
+	/** Resolves when the run answers the request, or when the hold runs out. */
+	readonly outcome: Promise<WebhookRunOutcome>;
+	/**
+	 * Stops the wait. The registry releases the wait by itself when the outcome
+	 * settles, so call this only to give up early. It is safe to call again.
+	 */
+	release(): void;
+}
 
 /** A request that is still open, and the subscription that feeds its answer. */
 type PendingWebhook = {
-	response: PendingWebhookResponse;
-	unsubscribe: UnsubscribeExecutionResponse;
+	executionId: ExecutionIdV2;
+	expectation: ResponseExpectation;
+	answer: IDeferredPromise<WebhookRunOutcome>;
+	/** Held so an answered run does not leave a timer behind for the whole hold. */
+	timeoutTimer: NodeJS.Timeout;
+	/** Set when the subscription is ready. */
+	unsubscribe?: UnsubscribeExecutionResponse;
+	released: boolean;
 };
 
 /**
@@ -28,14 +47,15 @@ export const MAX_PENDING_WEBHOOKS = 5000;
 export const SUBSCRIBE_TIMEOUT_MS = 30_000;
 
 /**
- * A service for hooking up responses from the data plane with webhook callers
- * expecting that response.
+ * Keeps track of the webhook requests that wait for a data-plane run. It owns
+ * the lifecycle of each wait: the capacity limit, the subscription, the
+ * response timeout, and the release.
  *
  * `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 {
+export class EngineV2WebhookResponseRegistry {
 	private receiver?: ExecutionResponseReceiver;
 
 	private readonly pendingWebhooks = new Map<string, PendingWebhook>();
@@ -71,7 +91,7 @@ export class EngineV2WebhookResponder {
 	async waitForResponse(
 		executionId: ExecutionIdV2,
 		expectation: ResponseExpectation,
-	): Promise<PendingWebhookResponse> {
+	): Promise<WebhookResponseWait> {
 		const { receiver } = this;
 		if (!receiver) {
 			throw new UnexpectedError('Engine v2 cannot wait for a response without a receiver');
@@ -90,21 +110,28 @@ export class EngineV2WebhookResponder {
 			});
 		}
 
-		const response = new PendingWebhookResponse({
+		const pending: PendingWebhook = {
 			executionId,
 			expectation,
-			timeoutMs: this.engineConfig.webhookResponseTimeout,
-			onRelease: (id) => this.release(id),
-		});
+			answer: createDeferredPromise<WebhookRunOutcome>(),
+			timeoutTimer: setTimeout(
+				() => this.settle(pending, { status: 'timeout' }),
+				this.engineConfig.webhookResponseTimeout,
+			).unref(),
+			released: false,
+		};
 
 		// Hold the slot before the subscription is ready, so requests that arrive
 		// meanwhile still count against the limit.
-		const pending: PendingWebhook = { response, unsubscribe: () => {} };
 		this.pendingWebhooks.set(executionId, pending);
 
-		pending.unsubscribe = await this.subscribe(receiver, response);
+		await this.subscribe(receiver, pending);
 
-		return response;
+		return {
+			executionId,
+			outcome: pending.answer.promise,
+			release: () => this.release(pending),
+		};
 	}
 
 	/**
@@ -116,29 +143,29 @@ export class EngineV2WebhookResponder {
 	 */
 	private async subscribe(
 		receiver: ExecutionResponseReceiver,
-		response: PendingWebhookResponse,
-	): Promise<UnsubscribeExecutionResponse> {
-		const { executionId } = response;
+		pending: PendingWebhook,
+	): Promise<void> {
+		const { executionId } = pending;
 		const subscription = receiver.receive(executionId, (received) =>
-			this.handle(received, response),
+			this.handle(received, pending),
 		);
-		let timeoutTimer: NodeJS.Timeout | undefined;
+		let subscribeTimeoutTimer: NodeJS.Timeout | undefined;
 		const timedOut = new Promise<'timed-out'>((resolve) => {
-			timeoutTimer = setTimeout(() => resolve('timed-out'), SUBSCRIBE_TIMEOUT_MS).unref();
+			subscribeTimeoutTimer = setTimeout(() => resolve('timed-out'), SUBSCRIBE_TIMEOUT_MS).unref();
 		});
 
 		let result: UnsubscribeExecutionResponse | 'timed-out';
 		try {
 			result = await Promise.race([subscription, timedOut]);
 		} catch (error) {
-			response.release();
+			this.release(pending);
 			throw error;
 		} finally {
-			clearTimeout(timeoutTimer);
+			clearTimeout(subscribeTimeoutTimer);
 		}
 
 		if (result === 'timed-out') {
-			response.release();
+			this.release(pending);
 			// The subscription can still complete. It must not outlive the request.
 			void subscription.then(
 				(unsubscribe) => unsubscribe(),
@@ -150,12 +177,23 @@ export class EngineV2WebhookResponder {
 			);
 		}
 
-		return result;
+		const unsubscribe = result;
+
+		// The wait can settle while the subscription is still pending.
+		if (pending.released) {
+			unsubscribe();
+			return;
+		}
+
+		pending.unsubscribe = unsubscribe;
 	}
 
-	private handle(received: ExecutionResponse, response: PendingWebhookResponse): void {
+	private handle(received: ExecutionResponse, pending: PendingWebhook): void {
+		if (pending.released) return;
+
 		try {
-			this.route(received, response);
+			const outcome = toWebhookOutcome(received, pending.expectation);
+			if (outcome) this.settle(pending, outcome);
 		} catch (error) {
 			this.logger.error('Failed to relay an engine v2 response', {
 				executionId: received.executionId,
@@ -165,46 +203,21 @@ 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;
+	/** The first outcome wins. Nothing more is needed from the run after it. */
+	private settle(pending: PendingWebhook, outcome: WebhookRunOutcome): void {
+		if (pending.released) return;
 
-			case 'response':
-				// A Buffer body arrives base64-encoded, because the channel is JSON.
-				response.resolveResponse(decodeBufferBody(received.payload));
-				return;
-
-			case 'ended':
-				this.onEnded(received, response);
-				return;
-		}
-	}
-
-	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,
-		});
+		pending.answer.resolve(outcome);
+		this.release(pending);
 	}
 
 	/** 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);
+	private release(pending: PendingWebhook): void {
+		if (pending.released) return;
+		pending.released = true;
+
+		clearTimeout(pending.timeoutTimer);
+		pending.unsubscribe?.();
+		this.pendingWebhooks.delete(pending.executionId);
 	}
 }

packages/cli/src/services/pending-webhook-response.ts

@@ -1,74 +0,0 @@
-import type { ResponseExpectation, 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: '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. */
-	| { status: 'undeliverable'; error: { name: string; message: string } }
-	| { status: 'timeout' };
-
-export type PendingWebhookResponseOptions = {
-	/** The run whose answer this handle carries. */
-	executionId: ExecutionIdV2;
-	/**
-	 * What the request waits for. Only `stepResponse` lets a Respond to Webhook
-	 * result answer it. The engine obeys this too, so this filter is a second
-	 * defense.
-	 */
-	expectation: ResponseExpectation;
-	/** 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 request gets an answer, or when the hold runs out. */
-	readonly settled: Promise<WebhookRunOutcome>;
-
-	readonly executionId: ExecutionIdV2;
-
-	private readonly expectation: ResponseExpectation;
-
-	private readonly onRelease: (executionId: string) => void;
-
-	private readonly answer = createDeferredPromise<WebhookRunOutcome>();
-
-	/** Held so an answered run does not leave a timer behind for the whole hold. */
-	private readonly timer: NodeJS.Timeout;
-
-	constructor({ executionId, expectation, timeoutMs, onRelease }: PendingWebhookResponseOptions) {
-		this.executionId = executionId;
-		this.expectation = expectation;
-		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);
-	}
-
-	resolveResponse(response: unknown): void {
-		if (this.expectation.kind === 'stepResponse') {
-			this.answer.resolve({ status: 'response', response });
-		}
-	}
-
-	release(): void {
-		clearTimeout(this.timer);
-		this.onRelease(this.executionId);
-	}
-}

packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts

@@ -63,7 +63,7 @@ import { type ResponseError } from '@n8n/errors';
 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 { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 import { OwnershipService } from '@/services/ownership.service';
 import { OAuth2FlowProxy } from '@/services/oauth2-flow-proxy.service';
 import type { ProtectedResource } from '@/services/protected-resource.registry';
@@ -2482,6 +2482,19 @@ describe('executeWebhook on engine v2', () => {
 		dataPlane = new Map();
 	});
 
+	/**
+	 * Makes the next wait time out, as the registry does when no answer arrives
+	 * within the response timeout.
+	 */
+	const timeOutNextWait = () =>
+		vi
+			.spyOn(Container.get(EngineV2WebhookResponseRegistry), 'waitForResponse')
+			.mockImplementationOnce(async (executionId) => ({
+				executionId,
+				outcome: Promise.resolve({ status: 'timeout' as const }),
+				release: vi.fn(),
+			}));
+
 	/** 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].engineV2Response?.executionId as string;
@@ -2495,9 +2508,9 @@ describe('executeWebhook on engine v2', () => {
 	};
 
 	beforeAll(() => {
-		// The host hands the responder its receiver at boot. Keeping each run's
+		// The host hands the response registry its receiver at boot. Keeping each run's
 		// handler is how a test plays the data plane answering.
-		Container.get(EngineV2WebhookResponder).useReceiver(
+		Container.get(EngineV2WebhookResponseRegistry).useReceiver(
 			mock<ExecutionResponseReceiver>({
 				receive: vi.fn(async (executionId: string, handler: (r: ExecutionResponse) => void) => {
 					dataPlane.set(executionId, handler);
@@ -2555,7 +2568,10 @@ describe('executeWebhook on engine v2', () => {
 		});
 
 		it('tells the dispatcher that the caller waits for the run end', async () => {
-			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			const waitForResponse = vi.spyOn(
+				Container.get(EngineV2WebhookResponseRegistry),
+				'waitForResponse',
+			);
 
 			await startWebhook({ responseMode: 'lastNode' });
 
@@ -2649,12 +2665,8 @@ describe('executeWebhook on engine v2', () => {
 		});
 
 		it('answers with a timeout when the run does not send an ended message', async () => {
-			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			timeOutNextWait();
 			const { responseCallback } = await startWebhook({ responseMode: 'lastNode' });
-			const pending = await 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([
@@ -2667,14 +2679,14 @@ describe('executeWebhook on engine v2', () => {
 		});
 
 		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 = await waitForResponse.mock.results[0]?.value;
+			const executionId = workflowRunner.run.mock.calls[0][0].engineV2Response
+				?.executionId as string;
 
-			expect(pending).toBeDefined();
-			pending?.resolve({
-				status: 'undeliverable',
-				error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+			dataPlane.get(executionId)?.({
+				type: 'undeliverable',
+				executionId,
+				error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
 			});
 
 			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
@@ -2690,7 +2702,10 @@ describe('executeWebhook on engine v2', () => {
 
 	describe('responseNode', () => {
 		it('tells the dispatcher that the caller waits for a step response', async () => {
-			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			const waitForResponse = vi.spyOn(
+				Container.get(EngineV2WebhookResponseRegistry),
+				'waitForResponse',
+			);
 
 			await startWebhook({ responseMode: 'responseNode' });
 
@@ -2780,12 +2795,8 @@ describe('executeWebhook on engine v2', () => {
 		});
 
 		it('answers with a timeout when no terminal outcome arrives', async () => {
-			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			timeOutNextWait();
 			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
-			const pending = await waitForResponse.mock.results[0]?.value;
-
-			expect(pending).toBeDefined();
-			pending?.resolve({ status: 'timeout' });
 
 			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
 			expect(responseCallback).toHaveBeenCalledWith(null, {

packages/cli/src/webhooks/engine-v2-webhooks.ts

@@ -24,7 +24,7 @@ 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';
+import type { WebhookRunOutcome } from '@/modules/engine-v2/webhook-response/webhook-outcome';
 
 /**
  * Trigger types the v2 path cannot serve. Each carries machinery the engine
@@ -95,7 +95,7 @@ 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
+		// Checked first: `EngineV2WebhookResponseRegistry.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()) {

packages/cli/src/webhooks/webhook-helpers.ts

@@ -80,8 +80,7 @@ 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 type { WebhookResponseWait } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
 import { WaitTracker } from '@/wait-tracker';
 import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
@@ -1036,7 +1035,7 @@ export async function executeWebhook(
 
 	/** Whether this run goes to the engine v2 data plane instead of the v1 path. */
 	let routesToEngineV2 = false;
-	let pendingEngineV2Response: PendingWebhookResponse | undefined;
+	let pendingEngineV2Response: WebhookResponseWait | undefined;
 	let runExecutionDataMerge = {};
 	const engineV2Webhooks = Container.get(EngineV2Webhooks);
 	let cleanupMultipartFiles: (() => Promise<void>) | undefined;
@@ -1242,10 +1241,13 @@ export async function executeWebhook(
 		// the run and the listener agree on it.
 		if (routesToEngineV2 && (responseMode === 'lastNode' || responseMode === 'responseNode')) {
 			const engineExecutionId = createExecutionIdV2();
-			pendingEngineV2Response = await Container.get(EngineV2WebhookResponder).waitForResponse(
-				engineExecutionId,
-				toResponseExpectation(responseMode),
+			// Loaded here, because only an engine v2 run needs the module code.
+			const { EngineV2WebhookResponseRegistry } = await import(
+				'@/modules/engine-v2/webhook-response/webhook-response-registry.service.js'
 			);
+			pendingEngineV2Response = await Container.get(
+				EngineV2WebhookResponseRegistry,
+			).waitForResponse(engineExecutionId, toResponseExpectation(responseMode));
 			runData.engineV2Response = { executionId: engineExecutionId, responseMode };
 		}
 
@@ -1346,66 +1348,62 @@ export async function executeWebhook(
 		 * 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 === 'response') {
-					if (!isHttpFullResponse(outcome.response)) {
-						throw new UnexpectedError('Engine v2 produced an invalid webhook response');
-					}
+		const waitForDataPlaneRun = async (waiting: WebhookResponseWait) => {
+			const outcome = await waiting.outcome;
+			if (outcome.status === 'response') {
+				if (!isHttpFullResponse(outcome.response)) {
+					throw new UnexpectedError('Engine v2 produced an invalid webhook response');
+				}
 
-					responder.markResponded();
-					await sendResponseNodeResponse(
-						outcome.response,
-						res,
-						responder,
-						workflowStartNode,
-						executionId,
-						workflow,
-					);
+				responder.markResponded();
+				await sendResponseNodeResponse(
+					outcome.response,
+					res,
+					responder,
+					workflowStartNode,
+					executionId,
+					workflow,
+				);
+				return undefined;
+			}
+
+			if (outcome.status === 'completed' || outcome.status === 'failed') {
+				if (outcome.status === 'completed' && responseMode === 'responseNode') {
+					responder.respondWith({ data: undefined, responseCode });
 					return undefined;
 				}
 
-				if (outcome.status === 'completed' || outcome.status === 'failed') {
-					if (outcome.status === 'completed' && responseMode === 'responseNode') {
-						responder.respondWith({ data: undefined, responseCode });
-						return undefined;
-					}
-
-					return await engineV2Webhooks.toRun(outcome, executionMode);
-				}
-
-				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 when the execution response later settles.
-				if (!responder.hasResponded) {
-					responder.respondWith({
-						data: { message: errorResponse.responseMessage },
-						responseCode: errorResponse.responseCode,
-					});
-				}
-
-				return undefined;
-			} finally {
-				waiting.release();
+				return await engineV2Webhooks.toRun(outcome, executionMode);
 			}
+
+			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 when the execution response later settles.
+			if (!responder.hasResponded) {
+				responder.respondWith({
+					data: { message: errorResponse.responseMessage },
+					responseCode: errorResponse.responseCode,
+				});
+			}
+
+			return undefined;
 		};
 
 		// Get a promise which resolves when the workflow did execute and send then response.

packages/cli/test/integration/engine-v2-webhooks.test.ts

@@ -21,7 +21,7 @@ import { CacheService } from '@n8n/backend-services';
 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 { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 import { Telemetry } from '@/telemetry';
 import { WebhookServer } from '@/webhooks/webhook-server';
 
@@ -77,11 +77,11 @@ beforeAll(async () => {
 		getExecution,
 		searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }),
 	});
-	// The host hands the responder its receiver at boot (`EngineV2Module.init`).
+	// The host hands the response registry 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(
+	Container.get(EngineV2WebhookResponseRegistry).useReceiver(
 		new InMemoryExecutionResponseReceiver(responseChannel, Container.get(Logger)),
 	);