diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts index 1250293ac77..85f38a3a361 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-response-hooks.test.ts @@ -1,5 +1,5 @@ import type { ResponseEmitter, StepExecutionRequest } from '@n8n/engine'; -import { ExecutionLifecycleHooks } from 'n8n-core'; +import { ENCODED_BUFFER_KEY, ExecutionLifecycleHooks } from 'n8n-core'; import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow'; import { describe, expect, it, vi } from 'vitest'; @@ -66,10 +66,42 @@ describe('attachResponseHooks', () => { expect(respond.send).toHaveBeenCalledOnce(); }); + describe('a Buffer body', () => { + const headers = { 'content-type': 'application/octet-stream', 'content-length': 5 }; + + it('is sent as a base64 envelope, with its headers and status code', async () => { + const { request, respond } = newRequest(); + const additionalData = newAdditionalData(); + + attachResponseHooks(additionalData, request); + await additionalData.hooks?.runHook('sendResponse', [ + { body: Buffer.from('hello'), headers, statusCode: 201 }, + ]); + + expect(respond.send).toHaveBeenCalledWith({ + body: { [ENCODED_BUFFER_KEY]: 'aGVsbG8=' }, + headers, + statusCode: 201, + }); + }); + + it('leaves the response the node produced untouched', async () => { + const { request } = newRequest(); + const additionalData = newAdditionalData(); + const body = Buffer.from('hello'); + const response = { body, headers, statusCode: 200 }; + + attachResponseHooks(additionalData, request); + await additionalData.hooks?.runHook('sendResponse', [response]); + + expect(response.body).toBe(body); + }); + }); + it.each([ - ['a buffer body', { body: Buffer.from('hi') }], ['a stream body', { body: { pipe: () => {} } }], ['a binary reference', { body: { binaryData: { id: 'file-1' } } }], + ['a bare Buffer in place of a response', Buffer.from('hi')], ])('refuses %s, which has no JSON form', async (_name, response) => { const { request } = newRequest(); const additionalData = newAdditionalData(); diff --git a/packages/@n8n/node-engine-compatibility/src/index.ts b/packages/@n8n/node-engine-compatibility/src/index.ts index 08a353cc210..eebe90abdeb 100644 --- a/packages/@n8n/node-engine-compatibility/src/index.ts +++ b/packages/@n8n/node-engine-compatibility/src/index.ts @@ -15,3 +15,4 @@ export type { StepDataLoader, V1StepExecutorDeps, } from './types'; +export { attachResponseHooks } from './v1-response-hooks'; diff --git a/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts b/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts index 4421d768952..1cfc7894e9b 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-response-hooks.ts @@ -1,5 +1,5 @@ import type { JsonValue, StepExecutionRequest } from '@n8n/engine'; -import { ExecutionLifecycleHooks } from 'n8n-core'; +import { encodeBufferBody, ExecutionLifecycleHooks } from 'n8n-core'; import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow'; import { UserError } from 'n8n-workflow'; @@ -52,16 +52,27 @@ export function attachResponseHooks( /** * A response crosses a process boundary as JSON, so anything without a JSON * form fails here rather than arriving silently mangled. + * + * A top-level Buffer body is base64-encoded into the envelope + * which the control plane decodes, the same one Engine v1 queue mode uses. + * Nested Buffers and streams are not supported, as in Engine v1 queue mode. */ function toJsonPayload(value: unknown): JsonValue { assertCarriable(value); - // Match Engine 1 queue mode. Nested Buffers and streams are not supported. - if (typeof value === 'object' && value !== null && 'body' in value) { + if (!hasBody(value)) return value as JsonValue; + + if (!Buffer.isBuffer(value.body)) { assertCarriable(value.body); + return value as JsonValue; } - return value as JsonValue; + // A shallow copy, so a host hook that runs after this one still sees the Buffer. + return encodeBufferBody({ ...value }) as JsonValue; +} + +function hasBody(value: unknown): value is { body: unknown } { + return typeof value === 'object' && value !== null && 'body' in value; } function assertCarriable(value: unknown): void { diff --git a/packages/cli/src/modules/engine-v2/__tests__/buffer-webhook-response.test.ts b/packages/cli/src/modules/engine-v2/__tests__/buffer-webhook-response.test.ts new file mode 100644 index 00000000000..a5feaccbfca --- /dev/null +++ b/packages/cli/src/modules/engine-v2/__tests__/buffer-webhook-response.test.ts @@ -0,0 +1,83 @@ +import type { Logger } from '@n8n/backend-common'; +import { mockLogger } from '@n8n/backend-test-utils'; +import type { EngineConfig } from '@n8n/config'; +import type { StepExecutionRequest } from '@n8n/engine'; +import { attachResponseHooks } from '@n8n/node-engine-compatibility'; +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 { 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. + * 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. + */ +describe('a Buffer webhook response through the response channel', () => { + const bytes = Buffer.from([0x00, 0xff, 0x10, 0x80]); + const headers = { 'content-type': 'image/png', 'content-length': bytes.length }; + + const buildPath = () => { + const channel = new InMemoryExecutionResponseChannel(); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); + const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); + + const responder = new EngineV2WebhookResponder( + mock({ webhookResponseTimeout: 50_000 }), + mock({ scoped: () => mock() }), + ); + responder.useReceiver(receiver); + + const executionId = createExecutionIdV2(); + const pending = responder.waitForResponse(executionId, true); + + const additionalData = {} as IWorkflowExecuteAdditionalData; + attachResponseHooks(additionalData, { + context: { + executionId, + stepId: 'step-1', + workflowId: 'wf-1', + mode: 'production', + iteration: 0, + callerContext: { hostMode: 'webhook' }, + }, + respond: sender.emitterFor(executionId), + } as unknown as StepExecutionRequest); + + return { additionalData, pending }; + }; + + it('delivers the original bytes, headers and status code to the control plane', async () => { + const { additionalData, pending } = buildPath(); + + await additionalData.hooks?.runHook('sendResponse', [ + { body: Buffer.from(bytes), headers, statusCode: 201 }, + ]); + + const outcome = await pending.settled; + expect(outcome).toEqual({ + status: 'response', + response: { body: bytes, headers, statusCode: 201 }, + }); + const { response } = outcome as { response: { body: unknown } }; + expect(Buffer.isBuffer(response.body)).toBe(true); + }); + + it('delivers a JSON body as before', async () => { + const { additionalData, pending } = buildPath(); + + await additionalData.hooks?.runHook('sendResponse', [ + { body: { ok: true }, headers: {}, statusCode: 200 }, + ]); + + await expect(pending.settled).resolves.toEqual({ + status: 'response', + response: { body: { ok: true }, headers: {}, statusCode: 200 }, + }); + }); +}); 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 ab69512723e..f7306d35935 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 @@ -1,5 +1,6 @@ import { mockLogger } from '@n8n/backend-test-utils'; import type { ExecutionResponse } from '@n8n/engine'; +import { ENCODED_BUFFER_KEY } from 'n8n-core'; import { InMemoryExecutionResponseChannel } from '../response-channel/in-memory-execution-response-channel'; import { InMemoryExecutionResponseReceiver } from '../response-channel/in-memory-execution-response-receiver'; @@ -82,6 +83,23 @@ describe('in-memory execution responses', () => { ); }); + it('delivers a base64 Buffer envelope unchanged', () => { + const channel = new InMemoryExecutionResponseChannel(); + const sender = new InMemoryExecutionResponseSender(channel, mockLogger()); + const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); + const seen: ExecutionResponse[] = []; + receiver.receive('exec-1', (response) => seen.push(response)); + const payload = { + body: { [ENCODED_BUFFER_KEY]: Buffer.from([0x00, 0xff, 0x10]).toString('base64') }, + headers: { 'content-type': 'application/octet-stream' }, + statusCode: 200, + }; + + sender.emitterFor('exec-1').send(payload); + + expect(seen).toEqual([{ type: 'response', executionId: 'exec-1', payload }]); + }); + it('validates a response before delivering it', () => { const channel = new InMemoryExecutionResponseChannel(); const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger()); 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 fb951bf9903..7772db8fde4 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 @@ -1,6 +1,7 @@ import type { Logger } from '@n8n/backend-common'; import type { EngineConfig } from '@n8n/config'; import type { ExecutionResponse } from '@n8n/engine'; +import { ENCODED_BUFFER_KEY } from 'n8n-core'; import { mock } from 'vitest-mock-extended'; import { createExecutionIdV2 } from '@/executions/execution-id'; @@ -151,6 +152,29 @@ describe('EngineV2WebhookResponder', () => { }); }); + it('restores a Buffer body the data plane sent as a base64 envelope', async () => { + const pending = responder.waitForResponse(createExecutionIdV2(), true); + 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 = responder.waitForResponse(createExecutionIdV2(), true); 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 1cacfee3a34..f763e1264b6 100644 --- a/packages/cli/src/services/engine-v2-webhook-responder.service.ts +++ b/packages/cli/src/services/engine-v2-webhook-responder.service.ts @@ -2,6 +2,7 @@ import { Logger } from '@n8n/backend-common'; import { EngineConfig } from '@n8n/config'; import { Service } from '@n8n/di'; import type { EndedMessage, ExecutionResponse } from '@n8n/engine'; +import { decodeBufferBody } from 'n8n-core'; import { OperationalError, UnexpectedError } from 'n8n-workflow'; import type { ExecutionIdV2 } from '@/executions/execution-id'; @@ -108,7 +109,8 @@ export class EngineV2WebhookResponder { return; case 'response': - response.resolveResponse(received.payload); + // A Buffer body arrives base64-encoded, because the channel is JSON. + response.resolveResponse(decodeBufferBody(received.payload)); return; case 'ended': diff --git a/packages/core/src/webhook-response/buffer-body-codec.ts b/packages/core/src/webhook-response/buffer-body-codec.ts index 131991fcd5d..63a91670913 100644 --- a/packages/core/src/webhook-response/buffer-body-codec.ts +++ b/packages/core/src/webhook-response/buffer-body-codec.ts @@ -10,13 +10,14 @@ import { BINARY_ENCODING } from 'n8n-workflow'; export const ENCODED_BUFFER_KEY = '__@N8nEncodedBuffer@__'; /** - * Replaces a Buffer body with its base64 envelope. Any other body passes through. + * Replaces a Buffer body with its base64 envelope. Every other value passes + * through, a payload that is not a full response included. * - * @param response Full response. Mutated and returned. + * @param response Response to send. Mutated and returned. * @returns The same `response`. */ -export function encodeBufferBody(response: T): T { - if (Buffer.isBuffer(response.body)) { +export function encodeBufferBody(response: T): T { + if (hasResponseBody(response) && Buffer.isBuffer(response.body)) { response.body = { [ENCODED_BUFFER_KEY]: response.body.toString(BINARY_ENCODING) }; }