From 68afc97ac99be3dabac8fd9ea183a1d66ce55588 Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Thu, 27 Aug 2026 13:53:18 +0000 Subject: [PATCH] feat(engine): Receive engine lifecycle events in the control plane (#37014) --- .../@n8n/config/src/configs/engine.config.ts | 12 ++ .../@n8n/config/test/engine.config.test.ts | 27 ++++ ...gine-control-plane-auth.middleware.test.ts | 118 ++++++++++++++ .../engine-control-plane-client.test.ts | 151 ++++++++++++++++++ .../engine-control-plane-server.test.ts | 142 ++++++++++++++++ .../engine-lifecycle-event.controller.test.ts | 110 +++++++++++++ .../__tests__/engine-v2.module.test.ts | 21 +++ .../__tests__/engine-v2.runtime.test.ts | 30 +++- .../engine-control-plane-auth.middleware.ts | 49 ++++++ .../engine-v2/engine-control-plane-client.ts | 60 +++++++ .../engine-v2/engine-control-plane-server.ts | 104 ++++++++++++ .../engine-lifecycle-event.controller.ts | 41 +++++ .../modules/engine-v2/engine-v2.constants.ts | 5 + .../src/modules/engine-v2/engine-v2.module.ts | 8 + .../modules/engine-v2/engine-v2.runtime.ts | 4 + 15 files changed, 881 insertions(+), 1 deletion(-) create mode 100644 packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-auth.middleware.test.ts create mode 100644 packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-client.test.ts create mode 100644 packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-server.test.ts create mode 100644 packages/cli/src/modules/engine-v2/__tests__/engine-lifecycle-event.controller.test.ts create mode 100644 packages/cli/src/modules/engine-v2/engine-control-plane-auth.middleware.ts create mode 100644 packages/cli/src/modules/engine-v2/engine-control-plane-client.ts create mode 100644 packages/cli/src/modules/engine-v2/engine-control-plane-server.ts create mode 100644 packages/cli/src/modules/engine-v2/engine-lifecycle-event.controller.ts create mode 100644 packages/cli/src/modules/engine-v2/engine-v2.constants.ts diff --git a/packages/@n8n/config/src/configs/engine.config.ts b/packages/@n8n/config/src/configs/engine.config.ts index e4af2c31e46..e6bc03fab32 100644 --- a/packages/@n8n/config/src/configs/engine.config.ts +++ b/packages/@n8n/config/src/configs/engine.config.ts @@ -50,4 +50,16 @@ export class EngineConfig { */ @Env('N8N_ENGINE_AUTH_SECRET', z.string().min(AUTH_SECRET_MIN_LENGTH)) authSecret: string = ''; + + /** Port the control plane server listens on. Its own, so it can be firewalled off from the editor API. */ + @Env('N8N_ENGINE_CONTROL_PLANE_PORT') + controlPlanePort: number = 3001; + + /** Bind address for the control plane server. Loopback by default: only a data plane calls it. */ + @Env('N8N_ENGINE_CONTROL_PLANE_HOST') + controlPlaneHost: string = '127.0.0.1'; + + /** Where the engine dials the control plane server. Defaults to loopback; set it when that is not reachable. */ + @Env('N8N_ENGINE_CONTROL_PLANE_BASE_URL') + controlPlaneBaseUrl: string = ''; } diff --git a/packages/@n8n/config/test/engine.config.test.ts b/packages/@n8n/config/test/engine.config.test.ts index 5da201e7176..f389e413efe 100644 --- a/packages/@n8n/config/test/engine.config.test.ts +++ b/packages/@n8n/config/test/engine.config.test.ts @@ -31,4 +31,31 @@ describe('EngineConfig', () => { expect(Container.get(EngineConfig).authSecret).toBe(''); expect(consoleWarnMock).toHaveBeenCalledWith(expect.stringContaining('N8N_ENGINE_AUTH_SECRET')); }); + + it('should leave the control plane base URL empty so the host picks the default', () => { + expect(Container.get(EngineConfig).controlPlaneBaseUrl).toBe(''); + }); + + it('should bind the control plane server to loopback by default', () => { + const config = Container.get(EngineConfig); + + expect(config.controlPlaneHost).toBe('127.0.0.1'); + expect(config.controlPlanePort).toBe(3001); + }); + + it('should read the control plane server bind address', () => { + process.env.N8N_ENGINE_CONTROL_PLANE_HOST = '0.0.0.0'; + process.env.N8N_ENGINE_CONTROL_PLANE_PORT = '4001'; + + const config = Container.get(EngineConfig); + + expect(config.controlPlaneHost).toBe('0.0.0.0'); + expect(config.controlPlanePort).toBe(4001); + }); + + it('should read the control plane base URL', () => { + process.env.N8N_ENGINE_CONTROL_PLANE_BASE_URL = 'http://cp.internal:5678'; + + expect(Container.get(EngineConfig).controlPlaneBaseUrl).toBe('http://cp.internal:5678'); + }); }); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-auth.middleware.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-auth.middleware.test.ts new file mode 100644 index 00000000000..dd718af5c92 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-auth.middleware.test.ts @@ -0,0 +1,118 @@ +import type { Logger } from '@n8n/backend-common'; +import type { EngineConfig } from '@n8n/config'; +import { ACTION_TOKEN, mintActionToken, mintIdentityToken } from '@n8n/engine'; +import type { NextFunction, Request, Response } from 'express'; +import jwt from 'jsonwebtoken'; +import type { Mocked } from 'vitest'; +import { mock } from 'vitest-mock-extended'; + +import { createEngineControlPlaneAuthMiddleware } from '../engine-control-plane-auth.middleware'; + +const authSecret = 'a'.repeat(32); + +describe('createEngineControlPlaneAuthMiddleware', () => { + let logger: Logger; + let next: NextFunction; + + const newResponse = () => { + const res = { status: vi.fn(), json: vi.fn() }; + res.status.mockReturnValue(res); + return res as unknown as Mocked; + }; + + const newRequest = (authorization?: string) => + ({ + method: 'POST', + originalUrl: '/internal/status-callback', + header: vi.fn((name: string) => + name.toLowerCase() === 'authorization' ? authorization : undefined, + ), + }) as unknown as Request; + + const authenticate = (authorization?: string, secret = authSecret) => { + const res = newResponse(); + const middleware = createEngineControlPlaneAuthMiddleware( + mock({ authSecret: secret }), + logger, + ); + middleware(newRequest(authorization), res, next); + return res; + }; + + const expectRejected = (res: Mocked) => { + expect(next).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(401); + expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('/internal/status-callback')); + }; + + beforeEach(() => { + logger = mock(); + next = vi.fn(); + }); + + it('accepts a callback token scoped to status writes', () => { + const res = authenticate(`Bearer ${mintActionToken(authSecret, 'lifecycle-events:write')}`); + + expect(next).toHaveBeenCalled(); + expect(res.status).not.toHaveBeenCalled(); + }); + + it('reads the secret per request, so one generated after startup still works', () => { + // The secret is set after construction. + const engineConfig = mock({ authSecret: '' }); + const middleware = createEngineControlPlaneAuthMiddleware(engineConfig, logger); + engineConfig.authSecret = authSecret; + + middleware( + newRequest(`Bearer ${mintActionToken(authSecret, 'lifecycle-events:write')}`), + newResponse(), + next, + ); + + expect(next).toHaveBeenCalled(); + }); + + it.each([ + ['no authorization header', undefined], + ['a non-bearer scheme', 'Basic abc'], + ['a token that is not a JWT', 'Bearer not-a-jwt'], + ])('rejects %s', (_label, authorization) => { + expectRejected(authenticate(authorization)); + }); + + it('rejects an identity token minted for the control plane to data plane direction', () => { + // Same secret; only the issuer and audience stop the replay. + const token = mintIdentityToken(authSecret, { cpId: 'cp-1', tenantId: 'tenant-1' }); + + expectRejected(authenticate(`Bearer ${token}`)); + }); + + it('rejects a token carrying a different scope', () => { + const token = jwt.sign({ scope: 'credential:read' }, authSecret, { + algorithm: 'HS256', + issuer: ACTION_TOKEN.issuer, + audience: ACTION_TOKEN.audience, + expiresIn: ACTION_TOKEN.ttlSeconds, + }); + + expectRejected(authenticate(`Bearer ${token}`)); + }); + + it('rejects a token signed with a different secret', () => { + const token = mintActionToken('b'.repeat(32), 'lifecycle-events:write'); + + expectRejected(authenticate(`Bearer ${token}`)); + }); + + it('rejects everything when no shared secret is configured', () => { + const token = mintActionToken(authSecret, 'lifecycle-events:write'); + + expectRejected(authenticate(`Bearer ${token}`, '')); + }); + + it('does not leak which check failed', () => { + const res = authenticate('Bearer not-a-jwt'); + + expect(res.json).toHaveBeenCalledWith({ code: 401, message: 'Unauthenticated' }); + }); +}); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-client.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-client.test.ts new file mode 100644 index 00000000000..0d977951f17 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-client.test.ts @@ -0,0 +1,151 @@ +import type { + HttpRequestClient, + HttpRequestClientOptions, + OutboundHttp, +} from '@n8n/backend-network'; +import type { EngineConfig } from '@n8n/config'; +import type { LifecycleEvent } from '@n8n/engine'; +import { InvalidActionTokenError, verifyActionToken } from '@n8n/engine'; +import { OperationalError } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import { EngineControlPlaneClient } from '../engine-control-plane-client'; + +const authSecret = 'a'.repeat(32); + +const events: LifecycleEvent[] = [ + { + type: 'execution:completed', + executionId: 'exec-1', + workflowId: 'wf-1', + at: '2026-08-24T10:00:00.000Z', + }, +]; + +describe('EngineControlPlaneClient', () => { + let http: HttpRequestClient; + let clientOptions: HttpRequestClientOptions | undefined; + let client: EngineControlPlaneClient; + let signal: AbortSignal; + + const respondWith = (statusCode: number) => { + vi.mocked(http.request).mockResolvedValue({ statusCode, body: '', headers: {} }); + }; + + /** Rebuilds the client so each test can vary the config. */ + const newClient = (engineConfig: Partial = {}) => { + http = mock(); + const outboundHttp = mock({ + requests: vi.fn((options?: HttpRequestClientOptions) => { + clientOptions = options; + return http; + }), + }); + + return new EngineControlPlaneClient( + mock({ + controlPlaneBaseUrl: '', + authSecret, + controlPlanePort: 3001, + ...engineConfig, + }), + outboundHttp, + ); + }; + + beforeEach(() => { + client = newClient(); + signal = new AbortController().signal; + }); + + describe('sendLifecycleEvents', () => { + it('posts the batch to the control plane status-callback endpoint', async () => { + respondWith(204); + + await client.sendLifecycleEvents(events, signal); + + expect(http.request).toHaveBeenCalledWith( + expect.objectContaining({ + url: '/internal/status-callback', + method: 'POST', + body: { events }, + json: true, + }), + ); + }); + + it('does not follow redirects, so the action token reaches only the configured host', async () => { + respondWith(204); + + await client.sendLifecycleEvents(events, signal); + + expect(http.request).toHaveBeenCalledWith( + expect.objectContaining({ disableFollowRedirect: true }), + ); + }); + + it('dials the control plane server on the loopback, not n8n main', () => { + expect(clientOptions?.baseURL).toBe('http://127.0.0.1:3001'); + }); + + it('dials the configured base URL when the control plane answers elsewhere', () => { + newClient({ controlPlaneBaseUrl: 'https://cp.internal:8443' }); + + expect(clientOptions?.baseURL).toBe('https://cp.internal:8443'); + }); + + it('opts out of SSRF protection for the n8n-controlled host', () => { + expect(clientOptions?.useDefaultSsrfPolicy).toBe('unsafe'); + }); + + it('leaves the send deadline to the engine, which owns it', () => { + // A client timeout would fire first and hide the engine's deadline. + expect(clientOptions?.timeout).toBeUndefined(); + }); + + it("forwards the engine's abort signal, so an abandoned batch cancels its request", async () => { + respondWith(204); + + await client.sendLifecycleEvents(events, signal); + + expect(http.request).toHaveBeenCalledWith(expect.objectContaining({ abortSignal: signal })); + }); + + it('mints a fresh token per request', () => { + expect(typeof clientOptions?.headers).toBe('function'); + }); + + it('sends an action token scoped to lifecycle-event writes that the control plane accepts', () => { + const headers = clientOptions?.headers; + const resolved = typeof headers === 'function' ? headers() : headers; + const authorization = resolved?.authorization ?? ''; + + expect(authorization).toMatch(/^Bearer .+/); + + const token = authorization.replace('Bearer ', ''); + + expect(() => verifyActionToken(authSecret, token, 'lifecycle-events:write')).not.toThrow(); + expect(() => verifyActionToken('b'.repeat(32), token, 'lifecycle-events:write')).toThrow( + InvalidActionTokenError, + ); + }); + + it.each([302, 400, 401, 500])( + 'rejects a batch the control plane answered %s', + async (statusCode) => { + respondWith(statusCode); + + await expect(client.sendLifecycleEvents(events, signal)).rejects.toThrow(OperationalError); + await expect(client.sendLifecycleEvents(events, signal)).rejects.toThrow( + String(statusCode), + ); + }, + ); + + it('resolves on a 204', async () => { + respondWith(204); + + await expect(client.sendLifecycleEvents(events, signal)).resolves.toBeUndefined(); + }); + }); +}); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-server.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-server.test.ts new file mode 100644 index 00000000000..fb3c7d34a8f --- /dev/null +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-control-plane-server.test.ts @@ -0,0 +1,142 @@ +import type { Logger } from '@n8n/backend-common'; +import type { EngineConfig } from '@n8n/config'; +import { mintActionToken, mintIdentityToken, type LifecycleEvent } from '@n8n/engine'; +import request from 'supertest'; +import { mock } from 'vitest-mock-extended'; + +import { EngineControlPlaneServer } from '../engine-control-plane-server'; +import { EngineLifecycleEventController } from '../engine-lifecycle-event.controller'; + +const authSecret = 'a'.repeat(32); + +const events: LifecycleEvent[] = [ + { + type: 'execution:completed', + executionId: 'exec-1', + workflowId: 'wf-1', + at: '2026-08-25T10:00:00.000Z', + }, +]; + +/** Binds for real, so these exercise the wiring rather than a mock app. */ +describe('EngineControlPlaneServer', () => { + let server: EngineControlPlaneServer; + let logger: Logger; + let serverLogger: Logger; + let baseUrl: string; + + const engineConfig = (overrides: Partial = {}) => + mock({ + authSecret, + controlPlaneHost: '127.0.0.1', + // Port 0: the OS picks a free one, so parallel files cannot clash. + controlPlanePort: 0, + ...overrides, + }); + + beforeEach(async () => { + logger = mock(); + serverLogger = mock(); + const controller = new EngineLifecycleEventController( + mock({ scoped: vi.fn().mockReturnValue(logger) }), + ); + server = new EngineControlPlaneServer( + engineConfig(), + controller, + mock({ scoped: vi.fn().mockReturnValue(serverLogger) }), + ); + await server.start(); + + baseUrl = `http://127.0.0.1:${server.port}`; + }); + + afterEach(async () => { + await server.stop(); + }); + + const post = (body: unknown, token?: string) => { + const req = request(baseUrl).post('/internal/status-callback'); + if (token) req.set('Authorization', `Bearer ${token}`); + return req.send(body as object); + }; + + it('logs the port it actually bound, not the configured one', () => { + // Configured as `0`, so the OS picked it: logging the configured value + // would tell an operator to dial port 0. + expect(serverLogger.info).toHaveBeenCalledWith( + expect.stringContaining(`http://127.0.0.1:${server.port}`), + ); + expect(serverLogger.info).not.toHaveBeenCalledWith(expect.stringContaining(':0')); + }); + + it('logs a server error that is not a failure to bind', () => { + // Nothing else handles these, so an unlogged one is a silent failure. + const error = Object.assign(new Error('boom'), { code: 'ECONNRESET' }); + + // @ts-expect-error reaching for the server's own error handler + server.server.emit('error', error); + + expect(serverLogger.error).toHaveBeenCalledWith('Engine 2.0 control plane server error', { + error, + }); + }); + + it('serves an open healthcheck', async () => { + const response = await request(baseUrl).get('/healthz'); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ status: 'ok' }); + }); + + it('accepts a batch from an authenticated data plane', async () => { + const response = await post({ events }, mintActionToken(authSecret, 'lifecycle-events:write')); + + expect(response.status).toBe(204); + expect(logger.debug).toHaveBeenCalledExactlyOnceWith( + 'Engine lifecycle event: execution:completed', + events[0], + ); + }); + + it.each([ + ['no token', undefined], + [ + 'an identity token minted for the other direction', + mintIdentityToken(authSecret, { + cpId: 'cp-1', + tenantId: 'tenant-1', + }), + ], + [ + 'a token signed with a different secret', + mintActionToken('b'.repeat(32), 'lifecycle-events:write'), + ], + ])('rejects %s', async (_label, token) => { + const response = await post({ events }, token); + + expect(response.status).toBe(401); + expect(response.body).toEqual({ code: 401, message: 'Unauthenticated' }); + }); + + it('rejects a batch the engine schema does not accept', async () => { + const response = await post( + { events: [{ type: 'nope' }] }, + mintActionToken(authSecret, 'lifecycle-events:write'), + ); + + expect(response.status).toBe(400); + }); + + it('does not authenticate the request body before the caller', async () => { + // An unauthenticated caller must not learn whether its body would validate. + const response = await post({ events: [{ type: 'nope' }] }); + + expect(response.status).toBe(401); + }); + + it('stops listening when stopped', async () => { + await server.stop(); + + await expect(request(baseUrl).get('/healthz')).rejects.toThrow(); + }); +}); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-lifecycle-event.controller.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-lifecycle-event.controller.test.ts new file mode 100644 index 00000000000..cc88a3ebe79 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-lifecycle-event.controller.test.ts @@ -0,0 +1,110 @@ +import type { Logger } from '@n8n/backend-common'; +import type { LifecycleEvent } from '@n8n/engine'; +import type { Request, Response } from 'express'; +import type { Mocked } from 'vitest'; +import { mock } from 'vitest-mock-extended'; + +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; + +import { EngineLifecycleEventController } from '../engine-lifecycle-event.controller'; + +const events: LifecycleEvent[] = [ + { + type: 'execution:started', + executionId: 'exec-1', + workflowId: 'wf-1', + mode: 'manual', + at: '2026-08-24T10:00:00.000Z', + }, + { + type: 'step:completed', + executionId: 'exec-1', + stepId: 'step-1', + nodeId: 'node-a', + nodeName: 'Edit Fields', + iteration: 0, + outputs: [[{ json: { greeting: 'hi' } }]], + at: '2026-08-24T10:00:01.000Z', + }, +]; + +describe('EngineLifecycleEventController', () => { + // The controller scopes its logger, so assert on the scoped one. + let logger: Logger; + let controller: EngineLifecycleEventController; + + const newResponse = () => { + const res = { + status: vi.fn(), + end: vi.fn(), + json: vi.fn(), + header: vi.fn(), + }; + res.status.mockReturnValue(res); + return res as unknown as Mocked; + }; + + const newRequest = (body: unknown = { events }) => ({ body }) as unknown as Request; + + beforeEach(() => { + logger = mock(); + controller = new EngineLifecycleEventController( + mock({ scoped: vi.fn().mockReturnValue(logger) }), + ); + }); + + describe('receiveLifecycleEvents', () => { + it('answers 204', async () => { + const res = newResponse(); + + await controller.receiveLifecycleEvents(newRequest(), res); + + expect(res.status).toHaveBeenCalledExactlyOnceWith(204); + expect(res.end).toHaveBeenCalled(); + }); + + it('logs every event in the batch, not just a count', async () => { + await controller.receiveLifecycleEvents(newRequest(), newResponse()); + + expect(logger.debug).toHaveBeenCalledTimes(2); + expect(logger.debug).toHaveBeenNthCalledWith( + 1, + 'Engine lifecycle event: execution:started', + events[0], + ); + }); + + it('logs a completed step by its output slot count, never its contents', async () => { + // A log must not become a copy of a user's execution data. + await controller.receiveLifecycleEvents(newRequest(), newResponse()); + + const [message, metadata] = vi.mocked(logger.debug).mock.calls[1]; + + expect(message).toBe('Engine lifecycle event: step:completed'); + expect(metadata).toEqual({ + type: 'step:completed', + executionId: 'exec-1', + stepId: 'step-1', + nodeId: 'node-a', + nodeName: 'Edit Fields', + iteration: 0, + at: '2026-08-24T10:00:01.000Z', + outputSlots: 1, + }); + expect(JSON.stringify(metadata)).not.toContain('greeting'); + }); + + it.each([ + ['a batch the engine schema rejects', { events: [{ type: 'nope' }] }], + ['an empty batch', { events: [] }], + ['a body that is not a batch at all', { hello: 'world' }], + ])('rejects %s without a reason', async (_label, body) => { + const res = newResponse(); + + await expect(controller.receiveLifecycleEvents(newRequest(body), res)).rejects.toThrow( + BadRequestError, + ); + expect(res.status).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts index 3e082f75245..6dedd1d28a2 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.module.test.ts @@ -4,6 +4,7 @@ import { Container } from '@n8n/di'; import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service'; +import { EngineControlPlaneServer } from '../engine-control-plane-server'; import { EngineDataPlaneClient } from '../engine-data-plane-client'; import { EngineV2Module } from '../engine-v2.module'; import { EngineV2Runtime } from '../engine-v2.runtime'; @@ -14,6 +15,7 @@ describe('EngineV2Module', () => { let engineConfig: EngineConfig; let runtime: EngineV2Runtime; let client: EngineDataPlaneClient; + let controlPlaneServer: EngineControlPlaneServer; beforeEach(() => { vi.clearAllMocks(); @@ -22,6 +24,7 @@ describe('EngineV2Module', () => { engineConfig = mockInstance(EngineConfig, { authSecret: '' }); runtime = mockInstance(EngineV2Runtime); client = mockInstance(EngineDataPlaneClient); + controlPlaneServer = mockInstance(EngineControlPlaneServer); Container.set(EngineDataPlaneProxyService, new EngineDataPlaneProxyService()); module = new EngineV2Module(); @@ -52,6 +55,15 @@ describe('EngineV2Module', () => { expect(client.startExecution).toHaveBeenCalledWith(request); }); + it('starts the control plane server before the engine, so a report always lands', async () => { + await module.init(); + + expect(controlPlaneServer.start).toHaveBeenCalled(); + expect(vi.mocked(controlPlaneServer.start).mock.invocationCallOrder[0]).toBeLessThan( + vi.mocked(runtime.init).mock.invocationCallOrder[0], + ); + }); + it('generates a secret when unset', async () => { await module.init(); @@ -73,5 +85,14 @@ describe('EngineV2Module', () => { expect(runtime.shutdown).toHaveBeenCalled(); }); + + it('stops the control plane server after the engine, so a final flush still lands', async () => { + await module.shutdown(); + + expect(controlPlaneServer.stop).toHaveBeenCalled(); + expect(vi.mocked(controlPlaneServer.stop).mock.invocationCallOrder[0]).toBeGreaterThan( + vi.mocked(runtime.shutdown).mock.invocationCallOrder[0], + ); + }); }); }); diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.runtime.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.runtime.test.ts index ba8dea838d0..0e056d51bf0 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/engine-v2.runtime.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-v2.runtime.test.ts @@ -4,6 +4,7 @@ import { mock } from 'vitest-mock-extended'; import type { NodeTypes } from '@/node-types'; +import type { EngineControlPlaneClient } from '../engine-control-plane-client'; import { EngineV2Runtime } from '../engine-v2.runtime'; // Hoisted so the `vi.mock` factories below, which vitest lifts above the imports, @@ -84,8 +85,10 @@ describe('EngineV2Runtime', () => { const nodeTypes = mock(); + let controlPlaneClient: EngineControlPlaneClient; + const newRuntime = (databaseUrl = 'postgres://engine') => - new EngineV2Runtime(engineConfig(databaseUrl), nodeTypes, mockLogger()); + new EngineV2Runtime(engineConfig(databaseUrl), nodeTypes, mockLogger(), controlPlaneClient); /** The `externalDependencies` callback the runtime handed to the engine. */ const externalDependencies = (stores: { executionStore: unknown; stepStore: unknown }) => { @@ -99,6 +102,7 @@ describe('EngineV2Runtime', () => { vi.clearAllMocks(); mocks.dataSource.isInitialized = false; mocks.listen.error = undefined; + controlPlaneClient = mock(); }); describe('init', () => { @@ -125,6 +129,30 @@ describe('EngineV2Runtime', () => { expect(mocks.engine.start).toHaveBeenCalled(); }); + it('injects a lifecycle event callback that reports to the control plane', async () => { + await newRuntime().init(); + + const events = [ + { + type: 'execution:completed' as const, + executionId: 'exec-1', + workflowId: 'wf-1', + at: '2026-08-24T10:00:00.000Z', + }, + ]; + const lifecycleEventCallback = externalDependencies({ executionStore: {}, stepStore: {} }) + .lifecycleEventCallback as (events: unknown[], signal: AbortSignal) => Promise; + const signal = new AbortController().signal; + + await lifecycleEventCallback(events, signal); + + // The signal rides along, so an abandoned batch cancels its request. + expect(controlPlaneClient.sendLifecycleEvents).toHaveBeenCalledExactlyOnceWith( + events, + signal, + ); + }); + it('injects the v1 step executor so v1-node steps can run', async () => { await newRuntime().init(); diff --git a/packages/cli/src/modules/engine-v2/engine-control-plane-auth.middleware.ts b/packages/cli/src/modules/engine-v2/engine-control-plane-auth.middleware.ts new file mode 100644 index 00000000000..b776d2def89 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/engine-control-plane-auth.middleware.ts @@ -0,0 +1,49 @@ +import type { Logger } from '@n8n/backend-common'; +import type { EngineConfig } from '@n8n/config'; +import { InvalidActionTokenError, verifyActionToken } from '@n8n/engine'; +import type { NextFunction, Request, Response } from 'express'; + +import { UnauthenticatedError } from '@/errors/response-errors/unauthenticated.error'; + +const BEARER_PREFIX = /^bearer /i; + +/** Express's `RequestHandler` may return a promise; verifying a token never does. */ +type SyncRequestHandler = (req: Request, res: Response, next: NextFunction) => void; + +/** + * Rejects a caller the shared secret does not vouch for. Reads the secret per + * request, because it is generated after this is constructed. + */ +export function createEngineControlPlaneAuthMiddleware( + engineConfig: EngineConfig, + logger: Logger, +): SyncRequestHandler { + const reject = (req: Request, res: Response, reason: string): void => { + // Logged, never returned: an operator needs the reason, a caller must not. + logger.warn(`Rejected ${req.method} ${req.originalUrl} - ${reason}`); + const error = new UnauthenticatedError(); + res.status(error.httpStatusCode).json({ code: error.errorCode, message: error.message }); + }; + + return (req, res, next) => { + const header = req.header('authorization') ?? ''; + if (!BEARER_PREFIX.test(header)) { + reject(req, res, 'no bearer token'); + return; + } + + try { + verifyActionToken( + engineConfig.authSecret, + header.replace(BEARER_PREFIX, ''), + 'lifecycle-events:write', + ); + } catch (error) { + if (!(error instanceof InvalidActionTokenError)) throw error; + reject(req, res, 'token rejected'); + return; + } + + next(); + }; +} diff --git a/packages/cli/src/modules/engine-v2/engine-control-plane-client.ts b/packages/cli/src/modules/engine-v2/engine-control-plane-client.ts new file mode 100644 index 00000000000..cda0ec7f876 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/engine-control-plane-client.ts @@ -0,0 +1,60 @@ +import type { HttpRequestClient } from '@n8n/backend-network'; +import { OutboundHttp } from '@n8n/backend-network'; +import { EngineConfig } from '@n8n/config'; +import { Service } from '@n8n/di'; +import type { LifecycleEvent } from '@n8n/engine'; +import { mintActionToken } from '@n8n/engine'; +import { OperationalError } from 'n8n-workflow'; + +import { STATUS_CALLBACK_PATH } from './engine-v2.constants'; + +/** + * Posts lifecycle events to the control plane server. Over HTTP even in-process, + * so the engine can move out of it without changing either side. + */ +@Service() +export class EngineControlPlaneClient { + private readonly http: HttpRequestClient; + + constructor( + private readonly engineConfig: EngineConfig, + outboundHttp: OutboundHttp, + ) { + this.http = outboundHttp.requests({ + // Fixed, n8n-controlled host. + useDefaultSsrfPolicy: 'unsafe', + // A bind address is not dialable, so default to loopback. + baseURL: + engineConfig.controlPlaneBaseUrl || `http://127.0.0.1:${engineConfig.controlPlanePort}`, + // A factory: each request needs a fresh token, and the secret is set later. + headers: () => ({ + authorization: `Bearer ${mintActionToken(this.engineConfig.authSecret, 'lifecycle-events:write')}`, + }), + }); + } + + /** Throws on a refused batch; the engine decides what a failed delivery costs. */ + async sendLifecycleEvents(events: LifecycleEvent[], signal: AbortSignal): Promise { + const response = await this.http.request({ + url: STATUS_CALLBACK_PATH, + method: 'POST', + body: { events }, + json: true, + returnFullResponse: true, + // Inspect the status here rather than catching a generic request error. + ignoreHttpStatusErrors: true, + // A redirect would forward the token to whatever host it names. + disableFollowRedirect: true, + // The engine owns the deadline and aborts on it. A client timeout would + // fire first and make that unreachable. + abortSignal: signal, + }); + + // 3xx too: redirects are not followed, so one is a misconfiguration. + if (response.statusCode >= 300) { + throw new OperationalError( + `Control plane refused a lifecycle event batch with ${response.statusCode}`, + ); + } + } +} diff --git a/packages/cli/src/modules/engine-v2/engine-control-plane-server.ts b/packages/cli/src/modules/engine-v2/engine-control-plane-server.ts new file mode 100644 index 00000000000..c6fed0d3b33 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/engine-control-plane-server.ts @@ -0,0 +1,104 @@ +import { inTest, Logger } from '@n8n/backend-common'; +import { EngineConfig } from '@n8n/config'; +import { Service } from '@n8n/di'; +import express, { type Application } from 'express'; +import { createServer, type Server } from 'node:http'; +import type { AddressInfo } from 'node:net'; + +import { bodyParser, rawBodyReader } from '@/middlewares'; +import { send } from '@/response-helper'; + +import { createEngineControlPlaneAuthMiddleware } from './engine-control-plane-auth.middleware'; +import { CONTROL_PLANE_PREFIX, STATUS_CALLBACK_PATH } from './engine-v2.constants'; +import { EngineLifecycleEventController } from './engine-lifecycle-event.controller'; + +/** + * Receives lifecycle events from the data plane. Its own server, not a route on + * n8n's main one, so this surface can be isolated from the editor API. + */ +@Service() +export class EngineControlPlaneServer { + private server: Server | undefined; + + /** The bound port, which differs from the configured one when that is `0`. */ + get port(): number | undefined { + return (this.server?.address() as AddressInfo | null)?.port; + } + + constructor( + private readonly engineConfig: EngineConfig, + private readonly lifecycleEventController: EngineLifecycleEventController, + private readonly logger: Logger, + ) { + this.logger = this.logger.scoped('engine-v2'); + } + + async start(): Promise { + const app = express(); + app.disable('x-powered-by'); + this.configureRoutes(app); + + const { controlPlaneHost: host, controlPlanePort: port } = this.engineConfig; + + this.server = createServer(app); + this.server.on('error', (error: NodeJS.ErrnoException) => { + if (error.code !== 'EADDRINUSE') { + // Nothing else handles these, so an unlogged one is a silent failure. + this.logger.error('Engine 2.0 control plane server error', { error }); + return; + } + + this.logger.error(`Engine 2.0 control plane port ${port} is already in use`); + // Skipped in tests, where exiting would kill the vitest worker. + if (!inTest) process.exit(1); + }); + + await new Promise((resolve, reject) => { + // Detached once listening: a later error is not a failed bind. + const onBindError = (error: Error) => reject(error); + this.server!.once('error', onBindError); + this.server!.listen(port, host, () => { + this.server!.off('error', onBindError); + resolve(); + }); + }); + + // An IPv6 literal needs brackets to read as a URL. + const shownHost = host.includes(':') ? `[${host}]` : host; + // The bound port, not the configured one, which is `0` when the OS picks it. + this.logger.info(`Engine 2.0 control plane listening on http://${shownHost}:${this.port}`); + } + + async stop(): Promise { + if (!this.server) return; + + await new Promise((resolve, reject) => { + this.server!.close((error) => (error ? reject(error) : resolve())); + }); + + // Dropped only on success, so a failed close is retried next time. + this.server = undefined; + } + + private configureRoutes(app: Application): void { + // Open: a liveness probe reveals nothing. + app.get('/healthz', (_req, res) => { + res.status(200).json({ status: 'ok' }); + }); + + // On the prefix, not the route, so a later route cannot forget either. + app.use( + CONTROL_PLANE_PREFIX, + createEngineControlPlaneAuthMiddleware(this.engineConfig, this.logger), + ); + // n8n's parser bounds the body by `N8N_PAYLOAD_SIZE_MAX`. + app.use(CONTROL_PLANE_PREFIX, rawBodyReader, bodyParser); + + app.post( + STATUS_CALLBACK_PATH, + send( + async (req, res) => await this.lifecycleEventController.receiveLifecycleEvents(req, res), + ), + ); + } +} diff --git a/packages/cli/src/modules/engine-v2/engine-lifecycle-event.controller.ts b/packages/cli/src/modules/engine-v2/engine-lifecycle-event.controller.ts new file mode 100644 index 00000000000..f62f176c7c7 --- /dev/null +++ b/packages/cli/src/modules/engine-v2/engine-lifecycle-event.controller.ts @@ -0,0 +1,41 @@ +import { Logger } from '@n8n/backend-common'; +import { Service } from '@n8n/di'; +import type { LifecycleEvent } from '@n8n/engine'; +import { lifecycleEventBatchSchema } from '@n8n/engine'; +import type { Request, Response } from 'express'; + +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; + +/** Handles `LifecycleEvent` batches from the engine 2.0 data plane. */ +@Service() +export class EngineLifecycleEventController { + constructor(private readonly logger: Logger) { + this.logger = this.logger.scoped('engine-v2'); + } + + async receiveLifecycleEvents(req: Request, res: Response): Promise { + const parsed = lifecycleEventBatchSchema.safeParse(req.body); + // The engine's own schema, so emitter and receiver cannot drift. The reason + // stays out of the response: only a data plane calls this. + if (!parsed.success) throw new BadRequestError('Invalid lifecycle event batch'); + + // TODO(CAT-2878): forward these to the editor and dispatch the error workflow. + for (const event of parsed.data.events) { + this.logger.debug(`Engine lifecycle event: ${event.type}`, toLogMetadata(event)); + } + + // Nothing to return, and re-delivery is harmless while this only logs. + res.status(204).end(); + } +} + +/** + * An event's identifiers, ready to log. Outputs become a slot count, so a log + * never becomes a copy of a user's execution data. + */ +function toLogMetadata(event: LifecycleEvent): Record { + if (event.type !== 'step:completed') return { ...event }; + + const { outputs, ...rest } = event; + return { ...rest, outputSlots: outputs.length }; +} diff --git a/packages/cli/src/modules/engine-v2/engine-v2.constants.ts b/packages/cli/src/modules/engine-v2/engine-v2.constants.ts new file mode 100644 index 00000000000..2f398d21fac --- /dev/null +++ b/packages/cli/src/modules/engine-v2/engine-v2.constants.ts @@ -0,0 +1,5 @@ +/** Prefix for every route the data plane calls. Auth and body limits mount on it. */ +export const CONTROL_PLANE_PREFIX = '/internal'; + +/** Where the data plane posts lifecycle event batches. Shared, so the two sides cannot disagree. */ +export const STATUS_CALLBACK_PATH = `${CONTROL_PLANE_PREFIX}/status-callback`; diff --git a/packages/cli/src/modules/engine-v2/engine-v2.module.ts b/packages/cli/src/modules/engine-v2/engine-v2.module.ts index 1cd36731c29..7ac535393fb 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.module.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.module.ts @@ -29,6 +29,10 @@ export class EngineV2Module implements ModuleInterface { engineConfig.authSecret = randomBytes(32).toString('hex'); } + // Before the engine, so nothing is reported with no server to receive it. + const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js'); + await Container.get(EngineControlPlaneServer).start(); + const { EngineV2Runtime } = await import('./engine-v2.runtime.js'); await Container.get(EngineV2Runtime).init(); @@ -45,5 +49,9 @@ export class EngineV2Module implements ModuleInterface { async shutdown() { const { EngineV2Runtime } = await import('./engine-v2.runtime.js'); await Container.get(EngineV2Runtime).shutdown(); + + // After the engine, so its final flush still has somewhere to land. + const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js'); + await Container.get(EngineControlPlaneServer).stop(); } } diff --git a/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts b/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts index 4bd7906b0d9..fb0771eceaf 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts @@ -14,6 +14,7 @@ import assert from 'node:assert'; import type { Server } from 'node:http'; import { NodeTypes } from '@/node-types'; +import { EngineControlPlaneClient } from './engine-control-plane-client'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; /** @@ -36,6 +37,7 @@ export class EngineV2Runtime { private readonly engineConfig: EngineConfig, private readonly nodeTypes: NodeTypes, private readonly logger: Logger, + private readonly controlPlaneClient: EngineControlPlaneClient, ) { this.logger = this.logger.scoped('engine-v2'); } @@ -83,6 +85,8 @@ export class EngineV2Runtime { admittance: new AllowAllAdmittance(), identityVerifier: new SharedSecretIdentityVerifier(this.engineConfig.authSecret), externalDependencies: ({ executionStore, stepStore }) => ({ + lifecycleEventCallback: async (events, signal) => + await this.controlPlaneClient.sendLifecycleEvents(events, signal), v1StepExecutor: new V1StepExecutor({ nodeTypes: this.nodeTypes, // TODO(CAT-2880): no credential access. A v1 node that needs credentials