diff --git a/packages/@n8n/config/src/configs/engine.config.ts b/packages/@n8n/config/src/configs/engine.config.ts index 00e3ce37203..c37fa5433df 100644 --- a/packages/@n8n/config/src/configs/engine.config.ts +++ b/packages/@n8n/config/src/configs/engine.config.ts @@ -82,7 +82,9 @@ export class EngineConfig { /** * How long (in ms) the control plane waits for a run to answer before it * answers the webhook request itself. The response channel is at-most-once, - * so a lost answer must not hold the request open forever. + * so a lost answer must not hold the request open forever. For a streaming + * request, each chunk starts the wait again, so this limits the silence + * between chunks, not the length of the stream. */ @Env('N8N_ENGINE_WEBHOOK_RESPONSE_TIMEOUT', positiveIntSchema.max(2147483647)) webhookResponseTimeout: number = 2 * 60 * 1000; diff --git a/packages/cli/src/__tests__/workflow-runner.test.ts b/packages/cli/src/__tests__/workflow-runner.test.ts index 58c1a93bc19..4d590935660 100644 --- a/packages/cli/src/__tests__/workflow-runner.test.ts +++ b/packages/cli/src/__tests__/workflow-runner.test.ts @@ -1762,8 +1762,15 @@ describe('pre-persist context establishment', () => { }); describe('streaming functionality', () => { - it('should setup heartbeat interval and sendChunk handler when streaming is enabled', async () => { + type StreamingResponse = Response & { flush: () => void }; + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should setup a v1 heartbeat interval and sendChunk handler when streaming is enabled', async () => { // ARRANGE + vi.useFakeTimers(); const activeExecutions = Container.get(ActiveExecutions); vi.spyOn(activeExecutions, 'add').mockResolvedValue('1'); vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValueOnce(); @@ -1771,7 +1778,7 @@ describe('streaming functionality', () => { const permissionChecker = Container.get(CredentialsPermissionChecker); vi.spyOn(permissionChecker, 'check').mockResolvedValueOnce(); - const mockResponse = mock({ writableEnded: false }); + const mockResponse = mock({ writableEnded: false }); const mockSetInterval = vi.spyOn(global, 'setInterval'); const data = mock({ @@ -1804,6 +1811,24 @@ describe('streaming functionality', () => { // sendChunk handler is still registered on lifecycle hooks expect(mockHooks.addHandler).toHaveBeenCalledWith('sendChunk', expect.any(Function)); + const closeHandler = mockResponse.once.mock.calls.find(([event]) => event === 'close')?.[1]; + closeHandler?.(); mockSetInterval.mockRestore(); }); + + it('does not manage the response lifecycle for a v2 run', async () => { + const dispatcher = Container.get(EngineV2Dispatcher); + vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(true); + vi.spyOn(dispatcher, 'start').mockResolvedValueOnce('dp-uuid'); + const setIntervalSpy = vi.spyOn(global, 'setInterval'); + + await runner.run( + mock({ + streamingEnabled: true, + httpResponse: mock({ writableEnded: false }), + }), + ); + + expect(setIntervalSpy).not.toHaveBeenCalled(); + }); }); diff --git a/packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts b/packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts index dd5a3ff08d4..5519081f267 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts @@ -60,6 +60,7 @@ describe('toWebhookOutcome', () => { expect(outcome).toEqual({ status: 'failed', + nodeId: 'c', nodeName: 'C', error: { name: 'NodeOperationError', message: 'it broke' }, }); @@ -110,7 +111,7 @@ describe('toWebhookOutcome', () => { expect(Buffer.isBuffer((outcome as { response: { body: unknown } }).response.body)).toBe(true); }); - it.each([runEnd, { kind: 'none' } as const])( + it.each([runEnd, { kind: 'stream' } as const, { kind: 'none' } as const])( 'ignores a Respond node result when the expectation is %j', (expectation) => { const payload = { body: { ignored: true }, headers: {}, statusCode: 200 }; diff --git a/packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts b/packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts index 8465d492c58..69391dd6cf9 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts @@ -11,11 +11,13 @@ import { MAX_PENDING_WEBHOOKS, SUBSCRIBE_TIMEOUT_MS, } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service'; +import type { ResponseStream } from '@/webhooks/streaming-webhook-response-heartbeat'; const TIMEOUT_MS = 50_000; const runEnd = { kind: 'runEnd' } as const; const stepResponse = { kind: 'stepResponse' } as const; +const stream = { kind: 'stream' } as const; /** * Stands in for the receiver. Only `src/modules/engine-v2/**` may import @@ -63,6 +65,10 @@ const newRegistry = (timeoutMs = TIMEOUT_MS) => mock({ scoped: () => mock() }), ); +const stillPending = Symbol('still pending'); +const raceWithPending = async (outcome: Promise) => + await Promise.race([outcome, Promise.resolve(stillPending)]); + describe('EngineV2WebhookResponseRegistry', () => { let deliver: (response: ExecutionResponse) => void; let registry: EngineV2WebhookResponseRegistry; @@ -79,10 +85,6 @@ describe('EngineV2WebhookResponseRegistry', () => { vi.useRealTimers(); }); - const stillPending = Symbol('still pending'); - const raceWithPending = async (outcome: Promise) => - await Promise.race([outcome, Promise.resolve(stillPending)]); - it('listens under the id the run is started with', async () => { const executionId = createExecutionIdV2(); @@ -230,6 +232,319 @@ describe('EngineV2WebhookResponseRegistry', () => { }); }); +describe('EngineV2WebhookResponseRegistry with a streaming request', () => { + let deliver: (response: ExecutionResponse) => void; + let registry: EngineV2WebhookResponseRegistry; + + beforeEach(() => { + vi.useFakeTimers(); + const fake = fakeReceiver(); + deliver = fake.deliver; + registry = newRegistry(); + registry.useReceiver(fake.receiver); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + const failedStep = { + status: 'failed', + lastStep: { + nodeId: 'c', + nodeName: 'C', + status: 'failed', + outputs: null, + error: { name: 'NodeOperationError', message: 'it broke' }, + }, + }; + + const openStream = () => mock({ writableEnded: false, destroyed: false }); + + const waitForStream = async (responseStream: ResponseStream) => + await registry.waitForResponse(createExecutionIdV2(), stream, responseStream); + + it('refuses to wait for a stream without a response stream', async () => { + await expect(registry.waitForResponse(createExecutionIdV2(), stream)).rejects.toThrow( + 'without a response stream', + ); + }); + + it.each([runEnd, stepResponse, { kind: 'none' } as const])( + 'does not write chunks when the expectation is %j', + async (expectation) => { + const responseStream = openStream(); + const pending = await registry.waitForResponse( + createExecutionIdV2(), + expectation, + responseStream, + ); + + deliver({ + type: 'chunk', + executionId: pending.executionId, + payload: { type: 'item', content: 'ignored' }, + }); + deliver(endedResponse(pending.executionId)); + await pending.outcome; + + expect(responseStream.write).not.toHaveBeenCalled(); + expect(responseStream.end).not.toHaveBeenCalled(); + }, + ); + + it('writes a chunk to the open response as one NDJSON line', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + const chunk = { type: 'item', content: 'hi' }; + deliver({ type: 'chunk', executionId: pending.executionId, payload: chunk }); + + expect(responseStream.write).toHaveBeenCalledWith(JSON.stringify(chunk) + '\n'); + expect(responseStream.flush).toHaveBeenCalledTimes(1); + pending.release(); + }); + + it('ignores a Respond node result', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + deliver({ + type: 'response', + executionId: pending.executionId, + payload: { body: { ignored: true }, headers: {}, statusCode: 200 }, + }); + deliver(endedResponse(pending.executionId)); + + await expect(pending.outcome).resolves.toMatchObject({ status: 'completed' }); + expect(responseStream.write).not.toHaveBeenCalled(); + }); + + it('ends a successful stream without an error chunk', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + deliver(endedResponse(pending.executionId)); + await pending.outcome; + + expect(responseStream.write).not.toHaveBeenCalled(); + expect(responseStream.end).toHaveBeenCalledTimes(1); + }); + + it('writes one fallback error chunk before a failed stream ends', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + deliver(endedResponse(pending.executionId, failedStep)); + await pending.outcome; + + expect(responseStream.write).toHaveBeenCalledTimes(1); + expect(responseStream.write).toHaveBeenCalledWith( + expect.stringContaining('"nodeId":"c","nodeName":"C"'), + ); + // The error message stays out of the stream. + expect(responseStream.write).toHaveBeenCalledWith( + expect.stringContaining('"content":"Workflow execution failed"'), + ); + expect(responseStream.write).not.toHaveBeenCalledWith(expect.stringContaining('it broke')); + expect(responseStream.flush).toHaveBeenCalledTimes(1); + expect(responseStream.write.mock.invocationCallOrder[0]).toBeLessThan( + responseStream.end.mock.invocationCallOrder[0], + ); + }); + + it('does not add a fallback error after the executor sent one', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + const errorChunk = { + type: 'error' as const, + content: 'it broke', + metadata: { nodeId: 'c', nodeName: 'C', runIndex: 0, itemIndex: 0, timestamp: 1 }, + }; + + deliver({ type: 'chunk', executionId: pending.executionId, payload: errorChunk }); + deliver(endedResponse(pending.executionId, failedStep)); + await pending.outcome; + + expect(responseStream.write).toHaveBeenCalledTimes(1); + expect(responseStream.write).toHaveBeenCalledWith(`${JSON.stringify(errorChunk)}\n`); + expect(responseStream.end).toHaveBeenCalledTimes(1); + }); + + it('writes and flushes an undeliverable response before it ends the stream', async () => { + const flush = vi.fn(); + const responseStream = mock({ writableEnded: false, flush }); + const pending = await waitForStream(responseStream); + + deliver({ + type: 'undeliverable', + executionId: pending.executionId, + error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + + await expect(pending.outcome).resolves.toEqual({ + status: 'undeliverable', + error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' }, + }); + expect(responseStream.write).toHaveBeenCalledWith( + expect.stringContaining('"nodeId":"unknown","nodeName":"unknown","runIndex":0,"itemIndex":0'), + ); + expect(flush).toHaveBeenCalledTimes(1); + expect(responseStream.end).toHaveBeenCalledTimes(1); + expect(flush.mock.invocationCallOrder[0]).toBeLessThan( + responseStream.end.mock.invocationCallOrder[0], + ); + }); + + it('writes an error chunk and ends the stream when the wait times out', async () => { + const impatient = newRegistry(1); + impatient.useReceiver(fakeReceiver().receiver); + const responseStream = openStream(); + + const pending = await impatient.waitForResponse(createExecutionIdV2(), stream, responseStream); + await vi.advanceTimersByTimeAsync(1); + + await expect(pending.outcome).resolves.toEqual({ status: 'timeout' }); + expect(responseStream.write).toHaveBeenCalledTimes(1); + expect(responseStream.write).toHaveBeenCalledWith( + expect.stringContaining('"type":"error","content":"Workflow execution timed out"'), + ); + expect(responseStream.end).toHaveBeenCalledTimes(1); + expect(responseStream.write.mock.invocationCallOrder[0]).toBeLessThan( + responseStream.end.mock.invocationCallOrder[0], + ); + }); + + it('measures the timeout from the last chunk', async () => { + const fake = fakeReceiver(); + const impatient = newRegistry(1_000); + impatient.useReceiver(fake.receiver); + const responseStream = openStream(); + const pending = await impatient.waitForResponse(createExecutionIdV2(), stream, responseStream); + const chunk = { + type: 'chunk' as const, + executionId: pending.executionId, + payload: { type: 'item', content: 'hi' }, + }; + + await vi.advanceTimersByTimeAsync(900); + fake.deliver(chunk); + await vi.advanceTimersByTimeAsync(900); + fake.deliver(chunk); + await vi.advanceTimersByTimeAsync(900); + + // 2.7 seconds in total, but never 1 second of silence. + await expect(raceWithPending(pending.outcome)).resolves.toBe(stillPending); + + await vi.advanceTimersByTimeAsync(100); + await expect(pending.outcome).resolves.toEqual({ status: 'timeout' }); + }); + + it('does not measure the timeout from chunks when the request does not wait for a stream', async () => { + const fake = fakeReceiver(); + const impatient = newRegistry(1_000); + impatient.useReceiver(fake.receiver); + const pending = await impatient.waitForResponse(createExecutionIdV2(), runEnd); + + await vi.advanceTimersByTimeAsync(900); + fake.deliver({ + type: 'chunk', + executionId: pending.executionId, + payload: { type: 'item', content: 'ignored' }, + }); + await vi.advanceTimersByTimeAsync(100); + + await expect(pending.outcome).resolves.toEqual({ status: 'timeout' }); + }); + + it('drops chunks after the run settles', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + deliver(endedResponse(pending.executionId)); + await pending.outcome; + deliver({ + type: 'chunk', + executionId: pending.executionId, + payload: { type: 'item', content: 'late' }, + }); + + expect(responseStream.write).not.toHaveBeenCalled(); + }); + + it('leaves the stream to the caller when the caller releases the wait', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + pending.release(); + + expect(responseStream.end).not.toHaveBeenCalled(); + }); + + describe('heartbeat', () => { + it('writes and flushes a keepalive after 30 seconds', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + await vi.advanceTimersByTimeAsync(30_000); + + expect(responseStream.write).toHaveBeenCalledWith('{"type":"keepalive"}\n'); + expect(responseStream.flush).toHaveBeenCalledTimes(1); + pending.release(); + }); + + it.each(['finish', 'close'] as const)('stops after the response emits %s', async (event) => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + const handler = responseStream.once.mock.calls.find( + ([registered]) => registered === event, + )?.[1]; + + handler?.(); + await vi.advanceTimersByTimeAsync(30_000); + + expect(responseStream.write).not.toHaveBeenCalled(); + pending.release(); + }); + + it('stops when the wait is released', async () => { + const responseStream = openStream(); + const pending = await waitForStream(responseStream); + + pending.release(); + await vi.advanceTimersByTimeAsync(30_000); + + expect(responseStream.write).not.toHaveBeenCalled(); + }); + + it('stops when the subscription fails', async () => { + const failing = newRegistry(); + failing.useReceiver({ + receive: async () => { + throw new Error('Redis is unavailable'); + }, + stop: async () => {}, + }); + const responseStream = openStream(); + + await expect( + failing.waitForResponse(createExecutionIdV2(), stream, responseStream), + ).rejects.toThrow('Redis is unavailable'); + await vi.advanceTimersByTimeAsync(30_000); + + expect(responseStream.write).not.toHaveBeenCalled(); + }); + + it('does not start for a request that does not wait for a stream', async () => { + const setIntervalSpy = vi.spyOn(global, 'setInterval'); + const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd); + + expect(setIntervalSpy).not.toHaveBeenCalled(); + pending.release(); + }); + }); +}); + describe('EngineV2WebhookResponseRegistry while the receiver subscribes', () => { /** A receiver whose subscriptions complete only when the test says so. */ function slowReceiver() { diff --git a/packages/cli/src/modules/engine-v2/webhook-response/streaming-response-writer.ts b/packages/cli/src/modules/engine-v2/webhook-response/streaming-response-writer.ts new file mode 100644 index 00000000000..ba2edb65a61 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/webhook-response/streaming-response-writer.ts @@ -0,0 +1,85 @@ +import type { JsonValue } from '@n8n/engine'; +import type { StructuredChunk } from 'n8n-workflow'; + +import { + type ResponseStream, + StreamingWebhookResponseHeartbeat, +} from '@/webhooks/streaming-webhook-response-heartbeat'; + +import type { WebhookRunOutcome } from './webhook-outcome'; + +/** + * Writes the NDJSON chunks of one run to its streaming webhook response. It + * keeps the stream open with a heartbeat, and it ends the stream when the run + * settles. + * + * It only writes. The response registry decides when the run is over, and + * when the writer stops. + */ +export class StreamingResponseWriter { + private deliveredError = false; + + private stopped = false; + + private readonly heartbeat: StreamingWebhookResponseHeartbeat; + + constructor(private readonly stream: ResponseStream) { + this.heartbeat = new StreamingWebhookResponseHeartbeat(stream); + } + + writeChunk(payload: JsonValue): void { + if (this.stopped) return; + + // The payload is opaque on the channel. Only this plane knows the v1 chunk. + const chunk = payload as unknown as StructuredChunk; + this.write(chunk); + if (chunk.type === 'error') this.deliveredError = true; + } + + /** Writes what the outcome needs the caller to know, and ends the stream. */ + finish(outcome: WebhookRunOutcome): void { + if (this.stopped) return; + + if (outcome.status === 'undeliverable') { + this.write(this.errorChunk(outcome.error.message)); + } + + if (outcome.status === 'timeout') { + // Timeout chunk to tell a cut-off stream apart from a complete one. + this.write(this.errorChunk('Workflow execution timed out')); + } + + if (outcome.status === 'failed' && !this.deliveredError) { + // The error message can hold request details, so the caller gets a + // generic text. The non-streaming modes do the same. + this.write(this.errorChunk('Workflow execution failed', outcome.nodeId, outcome.nodeName)); + } + + this.stop(); + if (!this.stream.writableEnded) this.stream.end(); + } + + /** + * Stops the writes and the heartbeat, but leaves the stream open. The caller + * that gives up on the run decides how the request ends. + */ + stop(): void { + if (this.stopped) return; + this.stopped = true; + this.heartbeat.stop(); + } + + private write(chunk: StructuredChunk): void { + this.stream.write(`${JSON.stringify(chunk)}\n`); + // The compression middleware buffers writes until it is flushed. + this.stream.flush?.(); + } + + private errorChunk(content: string, nodeId = 'unknown', nodeName = 'unknown'): StructuredChunk { + return { + type: 'error', + content, + metadata: { nodeId, nodeName, runIndex: 0, itemIndex: 0, timestamp: Date.now() }, + }; + } +} diff --git a/packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts b/packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts index 899df733711..1262cc7c281 100644 --- a/packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts +++ b/packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts @@ -9,7 +9,12 @@ export type LastNode = { nodeName: string; outputs: StepSlots }; export type WebhookRunOutcome = | { status: 'response'; response: unknown } | { status: 'completed'; lastNode?: LastNode } - | { status: 'failed'; nodeName: string; error?: { name: string; message: string } } + | { + status: 'failed'; + nodeId: string; + 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' }; @@ -43,13 +48,13 @@ export function toWebhookOutcome( return undefined; case 'ended': { - const { nodeName, outputs, error } = received.lastStep; + const { nodeId, 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 }; + return { status: 'failed', nodeId, nodeName, error }; case 'completed': return { diff --git a/packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts b/packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts index 10b3b42b426..608d54c4df8 100644 --- a/packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts +++ b/packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts @@ -6,11 +6,13 @@ import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise import { OperationalError, UnexpectedError } from 'n8n-workflow'; import type { ExecutionIdV2 } from '@/executions/execution-id'; +import type { ResponseStream } from '@/webhooks/streaming-webhook-response-heartbeat'; import type { ExecutionResponseReceiver, UnsubscribeExecutionResponse, } from '../response-channel/execution-response-receiver'; +import { StreamingResponseWriter } from './streaming-response-writer'; import { toWebhookOutcome, type WebhookRunOutcome } from './webhook-outcome'; /** The caller's view of an open webhook request that waits for its run. */ @@ -30,10 +32,15 @@ type PendingWebhook = { executionId: ExecutionIdV2; expectation: ResponseExpectation; answer: IDeferredPromise; - /** Held so an answered run does not leave a timer behind for the whole hold. */ + /** + * Held so an answered run does not leave a timer behind for the whole hold. + * A streaming request refreshes it on each chunk. + */ timeoutTimer: NodeJS.Timeout; /** Set when the subscription is ready. */ unsubscribe?: UnsubscribeExecutionResponse; + /** Set only when the request waits for a stream. */ + writer?: StreamingResponseWriter; released: boolean; }; @@ -49,7 +56,8 @@ export const SUBSCRIBE_TIMEOUT_MS = 30_000; /** * 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. + * response timeout, and the release. For a streaming request, it passes the + * chunks to a `StreamingResponseWriter`. * * `waitForResponse` should be called before starting the execution in case a * fast execution sends a response before we're listening for it. @@ -83,14 +91,18 @@ export class EngineV2WebhookResponseRegistry { * `StartExecution`. * @param expectation What the request waits for. Pass the same value to * `StartExecution`. - * @throws {UnexpectedError} If the execution response receiver is not set, or - * if the service already waits for this execution. + * @param responseStream The HTTP response to write chunks to. The service + * uses it only when the expectation is `stream`, and then it is required. + * @throws {UnexpectedError} If the execution response receiver is not set, + * if the service already waits for this execution, or if a stream is + * expected without a response stream. * @throws {OperationalError} If the service is at capacity, or if it cannot * listen for the response within `SUBSCRIBE_TIMEOUT_MS`. */ async waitForResponse( executionId: ExecutionIdV2, expectation: ResponseExpectation, + responseStream?: ResponseStream, ): Promise { const { receiver } = this; if (!receiver) { @@ -110,6 +122,12 @@ export class EngineV2WebhookResponseRegistry { }); } + if (expectation.kind === 'stream' && !responseStream) { + throw new UnexpectedError('Engine v2 cannot stream a response without a response stream', { + extra: { executionId }, + }); + } + const pending: PendingWebhook = { executionId, expectation, @@ -119,6 +137,10 @@ export class EngineV2WebhookResponseRegistry { this.engineConfig.webhookResponseTimeout, ).unref(), released: false, + writer: + expectation.kind === 'stream' && responseStream + ? new StreamingResponseWriter(responseStream) + : undefined, }; // Hold the slot before the subscription is ready, so requests that arrive @@ -192,6 +214,12 @@ export class EngineV2WebhookResponseRegistry { if (pending.released) return; try { + if (received.type === 'chunk' && pending.writer) { + pending.writer.writeChunk(received.payload); + // A chunk shows that the run is alive + pending.timeoutTimer.refresh(); + } + const outcome = toWebhookOutcome(received, pending.expectation); if (outcome) this.settle(pending, outcome); } catch (error) { @@ -207,6 +235,16 @@ export class EngineV2WebhookResponseRegistry { private settle(pending: PendingWebhook, outcome: WebhookRunOutcome): void { if (pending.released) return; + try { + pending.writer?.finish(outcome); + } catch (error) { + // A closed connection must not keep the request from settling. + this.logger.error('Failed to end an engine v2 webhook stream', { + executionId: pending.executionId, + error, + }); + } + pending.answer.resolve(outcome); this.release(pending); } @@ -217,6 +255,7 @@ export class EngineV2WebhookResponseRegistry { pending.released = true; clearTimeout(pending.timeoutTimer); + pending.writer?.stop(); pending.unsubscribe?.(); this.pendingWebhooks.delete(pending.executionId); } diff --git a/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts b/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts index 365a67f96dc..30844588986 100644 --- a/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts +++ b/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts @@ -13,7 +13,10 @@ const engineV2Webhooks = new EngineV2Webhooks( ); describe('EngineV2Webhooks.assertSupported', () => { - const assertSupported = (responseMode: WebhookResponseMode) => { + const assertSupported = ( + responseMode: WebhookResponseMode, + parameters: INode['parameters'] = {}, + ) => { const proxy = mock(); proxy.isAvailable.mockReturnValue(true); const webhooks = new EngineV2Webhooks(mock(), proxy); @@ -22,7 +25,7 @@ describe('EngineV2Webhooks.assertSupported', () => { workflowStartNode: mock({ name: 'Webhook', type: 'n8n-nodes-base.webhook', - parameters: {}, + parameters, }), responseMode, executionId: undefined, @@ -31,12 +34,28 @@ describe('EngineV2Webhooks.assertSupported', () => { // The engine mode no longer reaches this check, so these modes pass with a // remote data plane and in-process alike. - it.each(['onReceived', 'lastNode', 'responseNode'] as const)( + it.each(['onReceived', 'lastNode', 'responseNode', 'streaming'] as const)( 'allows %s responses in every engine mode', (responseMode) => { expect(() => assertSupported(responseMode)).not.toThrow(); }, ); + + // The streaming node sends its headers before it outputs the raw body, so a + // later refusal could not answer with a 400. + it.each([true, '={{ true }}'])('refuses a streaming response with rawBody %s', (rawBody) => { + expect(() => assertSupported('streaming', { options: { rawBody } })).toThrow( + 'Engine v2 cannot stream a response for the "Webhook" trigger with the Raw Body option yet.', + ); + }); + + it('allows a streaming response with rawBody turned off', () => { + expect(() => assertSupported('streaming', { options: { rawBody: false } })).not.toThrow(); + }); + + it('allows the raw body when the response does not stream', () => { + expect(() => assertSupported('lastNode', { options: { rawBody: true } })).not.toThrow(); + }); }); describe('EngineV2Webhooks.toRun', () => { @@ -77,6 +96,7 @@ describe('EngineV2Webhooks.toRun', () => { const run = await engineV2Webhooks.toRun( { status: 'failed', + nodeId: 'broken-node', nodeName: 'Broken node', error: { name: 'Error', message: 'The node failed' }, }, diff --git a/packages/cli/src/webhooks/__tests__/streaming-webhook-response-heartbeat.test.ts b/packages/cli/src/webhooks/__tests__/streaming-webhook-response-heartbeat.test.ts new file mode 100644 index 00000000000..1c14e0af751 --- /dev/null +++ b/packages/cli/src/webhooks/__tests__/streaming-webhook-response-heartbeat.test.ts @@ -0,0 +1,120 @@ +import { EventEmitter } from 'node:events'; + +import { + type ResponseStream, + StreamingWebhookResponseHeartbeat, +} from '@/webhooks/streaming-webhook-response-heartbeat'; + +const INTERVAL_MS = 30_000; +const KEEPALIVE = '{"type":"keepalive"}\n'; + +/** An HTTP response stand-in that emits `finish` and `close` like the real one. */ +class FakeResponseStream extends EventEmitter implements ResponseStream { + writableEnded = false; + + destroyed = false; + + readonly write = vi.fn(); + + readonly end = vi.fn(); + + readonly flush = vi.fn(); +} + +describe('StreamingWebhookResponseHeartbeat', () => { + let stream: FakeResponseStream; + + beforeEach(() => { + vi.useFakeTimers(); + stream = new FakeResponseStream(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('writes and flushes a keepalive at each interval', () => { + const heartbeat = new StreamingWebhookResponseHeartbeat(stream); + + vi.advanceTimersByTime(INTERVAL_MS - 1); + expect(stream.write).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1); + expect(stream.write).toHaveBeenCalledWith(KEEPALIVE); + expect(stream.flush).toHaveBeenCalledTimes(1); + + vi.advanceTimersByTime(INTERVAL_MS); + expect(stream.write).toHaveBeenCalledTimes(2); + + heartbeat.stop(); + }); + + it('writes a keepalive to a stream without a flush method', () => { + const withoutFlush: ResponseStream = { + writableEnded: false, + write: vi.fn(), + end: vi.fn(), + once: vi.fn(), + off: vi.fn(), + }; + const heartbeat = new StreamingWebhookResponseHeartbeat(withoutFlush); + + vi.advanceTimersByTime(INTERVAL_MS); + + expect(withoutFlush.write).toHaveBeenCalledWith(KEEPALIVE); + heartbeat.stop(); + }); + + it('writes nothing after stop', () => { + const heartbeat = new StreamingWebhookResponseHeartbeat(stream); + + heartbeat.stop(); + vi.advanceTimersByTime(INTERVAL_MS * 2); + + expect(stream.write).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each(['finish', 'close'] as const)('stops when the response emits %s', (event) => { + new StreamingWebhookResponseHeartbeat(stream); + + stream.emit(event); + vi.advanceTimersByTime(INTERVAL_MS); + + expect(stream.write).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('removes its listeners when it stops', () => { + const heartbeat = new StreamingWebhookResponseHeartbeat(stream); + expect(stream.listenerCount('finish')).toBe(1); + expect(stream.listenerCount('close')).toBe(1); + + heartbeat.stop(); + + expect(stream.listenerCount('finish')).toBe(0); + expect(stream.listenerCount('close')).toBe(0); + }); + + it.each([ + ['ended', { writableEnded: true }], + ['destroyed', { destroyed: true }], + ])('stops without a write when the response is %s', (_, state) => { + new StreamingWebhookResponseHeartbeat(stream); + Object.assign(stream, state); + + vi.advanceTimersByTime(INTERVAL_MS); + + expect(stream.write).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + expect(stream.listenerCount('close')).toBe(0); + }); + + it('allows a second stop', () => { + const heartbeat = new StreamingWebhookResponseHeartbeat(stream); + + heartbeat.stop(); + + expect(() => heartbeat.stop()).not.toThrow(); + }); +}); diff --git a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts index 1c7f79d9ebf..c608e490f42 100644 --- a/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts +++ b/packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts @@ -1,6 +1,7 @@ import { Logger } from '@n8n/backend-common'; import { EventService } from '@n8n/backend-services'; import { mockInstance } from '@n8n/backend-test-utils'; +import { EngineConfig } from '@n8n/config'; import type { Project, User } from '@n8n/db'; import { UserRepository } from '@n8n/db'; import { Container } from '@n8n/di'; @@ -52,6 +53,7 @@ import { SEND_AND_WAIT_OPERATION, createRunExecutionData, ExpressionError, + OperationalError, UserError, } from 'n8n-workflow'; import type { Readable } from 'stream'; @@ -2388,6 +2390,7 @@ describe('executeWebhook on engine v2', () => { const errorReporter = Container.get(ErrorReporter); /** Response handlers registered by the responder, by execution ID. */ let dataPlane: Map void>; + let unsubscribe = vi.fn<() => void>(); /** The data plane mints uuidv7 ids, not the numeric ids v1 uses. */ const ENGINE_EXECUTION_ID = '019606a1-0000-7000-8000-000000000001'; /** What the webhook node produces for a request that uploaded a file. */ @@ -2415,6 +2418,7 @@ describe('executeWebhook on engine v2', () => { executionId, destinationNode, responseHeaders, + response = mock({ headersSent: false, writableEnded: false }), }: { responseMode?: string; startNode?: INode; @@ -2423,6 +2427,7 @@ describe('executeWebhook on engine v2', () => { destinationNode?: IDestinationNode; /** The response headers configured on the webhook node, as an expression. */ responseHeaders?: string; + response?: express.Response; } = {}) => { webhookService.runWebhook.mockResolvedValue(webhookResult); @@ -2473,12 +2478,12 @@ describe('executeWebhook on engine v2', () => { }), executionId, mock({ method: 'POST', contentType: undefined, headers: {} }), - mock({ headersSent: false }), + response, responseCallback, destinationNode, ); - return { responseCallback, returned }; + return { response, responseCallback, returned }; }; beforeEach(() => { @@ -2497,6 +2502,7 @@ describe('executeWebhook on engine v2', () => { engineDataPlaneProxy.isAvailable.mockReturnValue(true); workflowRunner.run.mockResolvedValue(ENGINE_EXECUTION_ID); dataPlane = new Map(); + unsubscribe = vi.fn(); }); /** @@ -2531,7 +2537,8 @@ describe('executeWebhook on engine v2', () => { mock({ receive: vi.fn(async (executionId: string, handler: (r: ExecutionResponse) => void) => { dataPlane.set(executionId, handler); - return vi.fn(); + // Reassigned for each test, so read it when the listener goes. + return () => unsubscribe(); }), }), ); @@ -2617,11 +2624,15 @@ describe('executeWebhook on engine v2', () => { 'waitForResponse', ); - await startWebhook({ responseMode: 'lastNode' }); + const { response } = await startWebhook({ responseMode: 'lastNode' }); const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0]; expect(engineV2Response).toEqual({ responseMode: 'lastNode' }); - expect(waitForResponse).toHaveBeenCalledWith(engineV2ExecutionId, { kind: 'runEnd' }); + expect(waitForResponse).toHaveBeenCalledWith( + engineV2ExecutionId, + { kind: 'runEnd' }, + response, + ); }); it('answers with a 500 when a Respond node fails because the caller waits for the run end', async () => { @@ -2745,11 +2756,15 @@ describe('executeWebhook on engine v2', () => { 'waitForResponse', ); - await startWebhook({ responseMode: 'responseNode' }); + const { response } = await startWebhook({ responseMode: 'responseNode' }); const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0]; expect(engineV2Response).toEqual({ responseMode: 'responseNode' }); - expect(waitForResponse).toHaveBeenCalledWith(engineV2ExecutionId, { kind: 'stepResponse' }); + expect(waitForResponse).toHaveBeenCalledWith( + engineV2ExecutionId, + { kind: 'stepResponse' }, + response, + ); }); it('answers with the response published by the data plane', async () => { @@ -2836,6 +2851,112 @@ describe('executeWebhook on engine v2', () => { }); }); + describe('streaming', () => { + it('tells the dispatcher that the caller waits for a stream', async () => { + const waitForResponse = vi.spyOn( + Container.get(EngineV2WebhookResponseRegistry), + 'waitForResponse', + ); + + const { response } = await startWebhook({ responseMode: 'streaming' }); + + const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0]; + expect(engineV2Response).toEqual({ responseMode: 'streaming' }); + expect(waitForResponse).toHaveBeenCalledWith( + engineV2ExecutionId, + { kind: 'stream' }, + response, + ); + + // End the run, so its heartbeat and listener do not outlive the test. + answerRun('completed', { + nodeId: 'webhook-node-id', + nodeName: 'Webhook', + status: 'completed', + outputs: null, + }); + await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledTimes(1)); + }); + + it('ends the stream when the run cannot start', async () => { + // The streaming node sends the headers before the run starts. + const response = mock({ headersSent: true, writableEnded: false }); + workflowRunner.run.mockRejectedValueOnce(new Error('start failed')); + + await expect(startWebhook({ responseMode: 'streaming', response })).rejects.toThrow( + 'There was a problem executing the workflow', + ); + + expect(response.end).toHaveBeenCalledTimes(1); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + + it('ends the stream when the responder cannot listen for the run', async () => { + const response = mock({ headersSent: true, writableEnded: false }); + vi.spyOn( + Container.get(EngineV2WebhookResponseRegistry), + 'waitForResponse', + ).mockRejectedValueOnce(new OperationalError('Engine v2 is at capacity')); + + await expect(startWebhook({ responseMode: 'streaming', response })).rejects.toThrow( + 'There was a problem executing the workflow', + ); + + expect(response.end).toHaveBeenCalledTimes(1); + expect(workflowRunner.run).not.toHaveBeenCalled(); + }); + + it('completes with one error record when the failed run already sent one', async () => { + const { response } = await startWebhook({ responseMode: 'streaming' }); + const executionId = workflowRunner.run.mock.calls[0][0].engineV2ExecutionId as string; + expect(workflowRunner.run.mock.calls[0][0].httpResponse).toBeUndefined(); + const errorChunk = { + type: 'error' as const, + content: 'it broke', + metadata: { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + runIndex: 0, + itemIndex: 0, + timestamp: 1, + }, + }; + + dataPlane.get(executionId)?.({ type: 'chunk', executionId, payload: errorChunk }); + answerRun('failed', { + nodeId: 'edit-fields', + nodeName: 'Edit Fields', + status: 'failed', + outputs: null, + error: { name: 'NodeOperationError', message: 'it broke' }, + }); + + expect(response.write).toHaveBeenCalledTimes(1); + expect(response.write).toHaveBeenCalledWith(`${JSON.stringify(errorChunk)}\n`); + expect(response.flush).toHaveBeenCalledTimes(1); + expect(response.end).toHaveBeenCalledTimes(1); + }); + + it('ends an open response and releases its listener on timeout', async () => { + vi.useFakeTimers(); + const engineConfig = Container.get(EngineConfig); + const previousTimeout = engineConfig.webhookResponseTimeout; + engineConfig.webhookResponseTimeout = 10; + + try { + const { response, responseCallback } = await startWebhook({ responseMode: 'streaming' }); + await vi.advanceTimersByTimeAsync(10); + + expect(response.end).toHaveBeenCalledTimes(1); + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(responseCallback).not.toHaveBeenCalled(); + } finally { + engineConfig.webhookResponseTimeout = previousTimeout; + vi.useRealTimers(); + } + }); + }); + describe('rejections', () => { const reasonFrom = (responseCallback: ReturnType) => { const error = responseCallback.mock.calls[0][0] as ResponseError; @@ -2843,12 +2964,6 @@ describe('executeWebhook on engine v2', () => { }; it.each([ - { - name: 'the streaming response mode', - options: { responseMode: 'streaming' }, - message: - "Engine v2 does not support the 'streaming' response mode yet. Respond immediately instead.", - }, { name: 'the hostedChat response mode', options: { responseMode: 'hostedChat' }, diff --git a/packages/cli/src/webhooks/engine-v2-response-expectation.ts b/packages/cli/src/webhooks/engine-v2-response-expectation.ts index 269e119e7b8..0309e57c844 100644 --- a/packages/cli/src/webhooks/engine-v2-response-expectation.ts +++ b/packages/cli/src/webhooks/engine-v2-response-expectation.ts @@ -14,6 +14,8 @@ export function toResponseExpectation(responseMode?: EngineV2ResponseMode): Resp return { kind: 'runEnd' }; case 'responseNode': return { kind: 'stepResponse' }; + case 'streaming': + return { kind: 'stream' }; default: // `onReceived`, triggers, pollers and manual runs: nobody listens. return { kind: 'none' }; diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index 9fe689402f8..72b3e391dce 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -1,4 +1,5 @@ import { Service } from '@n8n/di'; +import { isRecord } from '@n8n/utils/is-record'; import type { INode, IRun, @@ -47,6 +48,7 @@ const SUPPORTED_RESPONSE_MODES = new Set([ 'onReceived', 'lastNode', 'responseNode', + 'streaming', ]); /** What the request says about a run, before the webhook node has produced anything. */ @@ -128,6 +130,22 @@ export class EngineV2Webhooks { `Engine v2 does not support the '${responseMode}' response mode yet. Respond immediately instead.`, ); } + + // With the raw body, a streaming Webhook node outputs a file after it sent + // the stream headers. `assertPayloadSupported` then cannot answer with a + // 400, so refuse the configuration here. + if (responseMode === 'streaming' && this.keepsRawBody(workflowStartNode)) { + throw new UserError( + `Engine v2 cannot stream a response for the "${workflowStartNode.name}" trigger with the Raw Body option yet.`, + ); + } + } + + private keepsRawBody(node: INode): boolean { + const options: unknown = node.parameters.options; + if (!isRecord(options)) return false; + // An expression can turn the option on, so only a missing or `false` value is safe. + return options.rawBody !== undefined && options.rawBody !== false; } /** Converts the data plane's answer to the shape the v1 response path reads. */ diff --git a/packages/cli/src/webhooks/streaming-webhook-response-heartbeat.ts b/packages/cli/src/webhooks/streaming-webhook-response-heartbeat.ts new file mode 100644 index 00000000000..4fa558a0e75 --- /dev/null +++ b/packages/cli/src/webhooks/streaming-webhook-response-heartbeat.ts @@ -0,0 +1,41 @@ +/** The HTTP response operations that a streaming webhook response uses. */ +export interface ResponseStream { + readonly writableEnded: boolean; + readonly destroyed?: boolean; + write(chunk: string): void; + end(): void; + flush?: () => void; + once(event: 'finish' | 'close', listener: () => void): void; + off(event: 'finish' | 'close', listener: () => void): void; +} + +const STREAMING_HEARTBEAT_INTERVAL_MS = 30_000; +const STREAMING_KEEPALIVE_CHUNK = '{"type":"keepalive"}\n'; + +/** Keeps a streaming webhook response open until it ends or disconnects. */ +export class StreamingWebhookResponseHeartbeat { + private stopped = false; + + private readonly timer: NodeJS.Timeout; + + constructor(private readonly stream: ResponseStream) { + this.timer = setInterval(() => { + if (this.stream.writableEnded || this.stream.destroyed) { + this.stop(); + return; + } + this.stream.write(STREAMING_KEEPALIVE_CHUNK); + this.stream.flush?.(); + }, STREAMING_HEARTBEAT_INTERVAL_MS); + this.stream.once('finish', this.stop); + this.stream.once('close', this.stop); + } + + readonly stop = () => { + if (this.stopped) return; + this.stopped = true; + clearInterval(this.timer); + this.stream.off('finish', this.stop); + this.stream.off('close', this.stop); + }; +} diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index 221b91d351f..92704c24d14 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -1232,7 +1232,8 @@ export async function executeWebhook( { executionId }, ); // TODO: Add check for streaming nodes here - runData.httpResponse = res; + // On engine v2, the response registry writes the stream instead. + if (!routesToEngineV2) runData.httpResponse = res; runData.streamingEnabled = true; // No `responder.respondWith()` here, unlike the formPage and hostedChat // branches: streaming requires the trigger to have taken over the @@ -1246,14 +1247,19 @@ export async function executeWebhook( if (engineV2ExecutionId !== undefined) { // Before the run, because a short workflow answers before `startExecution` // returns and nothing replays a missed response. - if (responseMode === 'lastNode' || responseMode === 'responseNode') { + if ( + responseMode === 'lastNode' || + responseMode === 'responseNode' || + responseMode === 'streaming' + ) { // 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' ); + // The registry writes to `res` only when the expectation is `stream`. pendingEngineV2Response = await Container.get( EngineV2WebhookResponseRegistry, - ).waitForResponse(engineV2ExecutionId, toResponseExpectation(responseMode)); + ).waitForResponse(engineV2ExecutionId, toResponseExpectation(responseMode), res); runData.engineV2Response = { responseMode }; } // The files the node stored are under this id, so the run must use it. @@ -1408,7 +1414,8 @@ export async function executeWebhook( ...(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. + // second response when the execution response later settles. A streaming + // request is marked as answered, and the registry ends its stream. if (!responder.hasResponded) { responder.respondWith({ data: { message: errorResponse.responseMessage }, @@ -1551,6 +1558,11 @@ export async function executeWebhook( } catch (e: unknown) { // Nothing will ever answer this one, so stop waiting for it. pendingEngineV2Response?.release(); + // The trigger already sent the stream headers, so no error response can + // follow. End the stream, or the caller waits for chunks that never come. + if (routesToEngineV2 && responseMode === 'streaming' && res.headersSent && !res.writableEnded) { + res.end(); + } const error = ensureError(e); let responseError: Error; diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index 0701c58e436..ed7cecfabdf 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -67,15 +67,10 @@ import type { PoolConfigService } from '@/scaling/pool-config.service.ee'; import type { ScalingService } from '@/scaling/scaling.service'; import type { Job, JobData } from '@/scaling/scaling.types'; import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service'; +import { StreamingWebhookResponseHeartbeat } from '@/webhooks/streaming-webhook-response-heartbeat'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service'; -/** Interval between keepalive writes on streaming responses to prevent proxy timeouts */ -const STREAMING_HEARTBEAT_INTERVAL_MS = 30_000; - -/** JSON chunk written periodically to keep the streaming connection alive through reverse proxies */ -const STREAMING_KEEPALIVE_CHUNK = '{"type":"keepalive"}\n'; - /** How long to keep rechecking the execution status after a max-stalled-count error before failing the run */ const MAX_STALLED_COUNT_GRACE_WINDOW_MS = 30 * Time.seconds.toMilliseconds; @@ -455,20 +450,9 @@ export class WorkflowRunner { if (responsePromise) { this.activeExecutions.attachResponsePromise(executionId, responsePromise); } - - // Set up streaming heartbeat on the main process that holds the HTTP response. - // This must happen BEFORE the queue/local decision because in queue mode the - // execution runs on a worker process that has no access to the HTTP response. - let heartbeatInterval: NodeJS.Timeout | undefined; - if (data.streamingEnabled === true && data.httpResponse) { - const res = data.httpResponse; - heartbeatInterval = setInterval(() => { - if (!res.writableEnded) { - res.write(STREAMING_KEEPALIVE_CHUNK); - flushResponse(res); - } - }, STREAMING_HEARTBEAT_INTERVAL_MS); - } + // The v1 main process owns the HTTP response. Start this before the + // queue/local decision because a worker cannot write to it. + const heartbeat = this.setupV1StreamingHeartbeat(data); // @TODO: Reduce to true branch once feature is stable const shouldEnqueue = @@ -500,7 +484,7 @@ export class WorkflowRunner { } catch (error) { // A failed start means the post-execute promise that normally clears the // heartbeat never settles, so clear it here. - if (heartbeatInterval) clearInterval(heartbeatInterval); + heartbeat?.stop(); throw error; } @@ -530,16 +514,23 @@ export class WorkflowRunner { } // Clean up the streaming heartbeat when the execution finishes - if (heartbeatInterval) { + if (heartbeat) { const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId); void postExecutePromise.finally(() => { - clearInterval(heartbeatInterval); + heartbeat.stop(); }); } return executionId; } + private setupV1StreamingHeartbeat( + data: IWorkflowExecutionDataProcess, + ): StreamingWebhookResponseHeartbeat | undefined { + if (data.streamingEnabled !== true || !data.httpResponse) return undefined; + return new StreamingWebhookResponseHeartbeat(data.httpResponse); + } + private resolvePinData(data: IWorkflowExecutionDataProcess): IPinData | undefined { if (['manual', 'evaluation'].includes(data.executionMode)) { return data.pinData ?? data.workflowData.pinData; diff --git a/packages/cli/test/integration/engine-v2-webhooks.test.ts b/packages/cli/test/integration/engine-v2-webhooks.test.ts index 6b8b03a0113..2a7386edc41 100644 --- a/packages/cli/test/integration/engine-v2-webhooks.test.ts +++ b/packages/cli/test/integration/engine-v2-webhooks.test.ts @@ -20,6 +20,7 @@ import { agent as testAgent } from 'supertest'; 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 { InMemoryExecutionResponseSender } from '@/modules/engine-v2/response-channel/in-memory-execution-response-sender'; import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; import { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service'; import { Telemetry } from '@/telemetry'; @@ -42,6 +43,7 @@ const getExecution = vi.fn(); let builder: User; let webhookAgent: SuperAgentTest; let webhookTestEndpoint: string; +let responseSender: InMemoryExecutionResponseSender; const webhookNode = (webhookId: string): INode => ({ id: randomUUID(), @@ -84,6 +86,7 @@ beforeAll(async () => { Container.get(EngineV2WebhookResponseRegistry).useReceiver( new InMemoryExecutionResponseReceiver(responseChannel, Container.get(Logger)), ); + responseSender = new InMemoryExecutionResponseSender(responseChannel, Container.get(Logger)); // `/webhook-test/*` is mounted only when a server opts into test webhooks. class EditorFacingWebhookServer extends WebhookServer { @@ -142,11 +145,40 @@ describe('webhook runs on engine v2', () => { expect(executions.filter((e) => e.workflowId === workflow.id)).toHaveLength(0); }); - test('answers 400 with the reason when the response mode is unsupported', async () => { + test('streams the response from the data plane', async () => { const webhookId = randomUUID(); const trigger = webhookNode(webhookId); trigger.parameters.responseMode = 'streaming'; const workflow = await createV2Workflow(trigger); + const chunk = { + type: 'item', + content: 'hello', + metadata: { + nodeId: trigger.id, + nodeName: trigger.name, + runIndex: 0, + itemIndex: 0, + timestamp: Date.now(), + }, + }; + + startExecution.mockImplementationOnce(async (request) => { + responseSender.send({ type: 'chunk', executionId: request.executionId, payload: chunk }); + responseSender.send({ + type: 'ended', + executionId: request.executionId, + workflowId: workflow.id, + status: 'completed', + lastStep: { + nodeId: trigger.id, + nodeName: trigger.name, + status: 'completed', + outputs: [], + }, + }); + + return { executionId: request.executionId }; + }); await startListening(workflow.id); @@ -154,8 +186,60 @@ describe('webhook runs on engine v2', () => { .post(`/${webhookTestEndpoint}/${webhookId}`) .send({ order: 42 }); - expect(response.statusCode).toBe(400); - expect(response.body.message).toContain("does not support the 'streaming' response mode yet"); - expect(startExecution).not.toHaveBeenCalled(); + expect(response.statusCode).toBe(200); + expect(response.text).toBe(`${JSON.stringify(chunk)}\n`); + expect(startExecution).toHaveBeenCalledWith( + expect.objectContaining({ responseExpectation: { kind: 'stream' } }), + ); + }); + + test('closes a failed stream without duplicating its error chunk', async () => { + const webhookId = randomUUID(); + const trigger = webhookNode(webhookId); + trigger.parameters.responseMode = 'streaming'; + const workflow = await createV2Workflow(trigger); + const errorChunk = { + type: 'error' as const, + content: 'it broke', + metadata: { + nodeId: trigger.id, + nodeName: trigger.name, + runIndex: 0, + itemIndex: 0, + timestamp: Date.now(), + }, + }; + + startExecution.mockImplementationOnce(async (request) => { + responseSender.send({ + type: 'chunk', + executionId: request.executionId, + payload: errorChunk, + }); + responseSender.send({ + type: 'ended', + executionId: request.executionId, + workflowId: workflow.id, + status: 'failed', + lastStep: { + nodeId: trigger.id, + nodeName: trigger.name, + status: 'failed', + outputs: null, + error: { name: 'NodeOperationError', message: 'it broke' }, + }, + }); + + return { executionId: request.executionId }; + }); + + await startListening(workflow.id); + + const response = await webhookAgent + .post(`/${webhookTestEndpoint}/${webhookId}`) + .send({ order: 42 }); + + expect(response.statusCode).toBe(200); + expect(response.text).toBe(`${JSON.stringify(errorChunk)}\n`); }); }); diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index ba0932bb9de..fc1354699ec 100644 --- a/packages/workflow/src/interfaces.ts +++ b/packages/workflow/src/interfaces.ts @@ -3802,7 +3802,7 @@ export interface IWorkflowExecutionDataProcess { * which answer the caller waits for. Without this field, nobody waits. */ engineV2Response?: { - responseMode: 'lastNode' | 'responseNode'; + responseMode: 'lastNode' | 'responseNode' | 'streaming'; }; startedAt?: Date;