feat(core): Run engine 2.0 as a standalone data plane (no-changelog) (#39440)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
mfsiega 2026-09-24 19:51:43 +00:00 committed by GitHub
parent e838224264
commit de238f3dde
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 595 additions and 89 deletions

View File

@ -158,6 +158,8 @@ jobs:
packages/@n8n/node-engine-compatibility/**
packages/@n8n/config/src/configs/engine.config.ts
packages/cli/src/modules/engine-v2/**
packages/cli/src/commands/engine.ts
packages/cli/src/commands/base-command.ts
packages/cli/src/services/engine-*
packages/cli/src/executions/**
packages/cli/src/webhooks/**

View File

@ -8,7 +8,7 @@ const InstanceRegistrationSchemaV1 = z
.string()
.regex(/^[a-zA-Z0-9\-_]+$/, 'instanceKey must be alphanumeric with hyphens/underscores'),
hostId: z.string(),
instanceType: z.enum(['main', 'worker', 'webhook']),
instanceType: z.enum(['main', 'worker', 'webhook', 'engine']),
instanceRole: z.enum(['leader', 'follower', 'unset']),
version: z.string(),
registeredAt: z.number(),
@ -33,7 +33,7 @@ const ClusterVersionMismatchSchema = z.object({
z.object({
instanceKey: z.string(),
hostId: z.string(),
instanceType: z.enum(['main', 'worker', 'webhook']),
instanceType: z.enum(['main', 'worker', 'webhook', 'engine']),
instanceRole: z.enum(['leader', 'follower', 'unset']),
}),
),

View File

@ -9,8 +9,20 @@ import { positiveIntSchema } from '../schemas';
*/
const AUTH_SECRET_MIN_LENGTH = 32;
const engineModeSchema = z.enum(['in-process', 'remote']);
export type EngineMode = z.infer<typeof engineModeSchema>;
@Config
export class EngineConfig {
/**
* Where the data plane runs. `in-process`: this main hosts it. `remote`:
* another process hosts it (`n8n engine`), and this main runs only the
* control plane side. Remote mode needs `N8N_ENGINE_BASE_URL` and
* `N8N_ENGINE_AUTH_SECRET`.
*/
@Env('N8N_ENGINE_MODE', engineModeSchema)
mode: EngineMode = 'in-process';
/** Port the engine HTTP server listens on. */
@Env('N8N_ENGINE_PORT')
port: number = 3000;

View File

@ -82,6 +82,7 @@ export {
SSRF_DEFAULT_BLOCKED_IP_RANGES,
} from './configs/ssrf-protection.config';
export { EngineConfig } from './configs/engine.config';
export type { EngineMode } from './configs/engine.config';
export { ExecutionsConfig } from './configs/executions.config';
export { ScalingModeConfig, WorkerPoolConfig } from './configs/scaling-mode.config';
export { LOG_SCOPES } from './configs/logging.config';

View File

@ -58,4 +58,21 @@ describe('EngineConfig', () => {
expect(Container.get(EngineConfig).controlPlaneBaseUrl).toBe('http://cp.internal:5678');
});
it('should host the data plane in-process by default', () => {
expect(Container.get(EngineConfig).mode).toBe('in-process');
});
it('should read the remote mode', () => {
process.env.N8N_ENGINE_MODE = 'remote';
expect(Container.get(EngineConfig).mode).toBe('remote');
});
it('should reject an unknown mode and fall back to in-process', () => {
process.env.N8N_ENGINE_MODE = 'sidecar';
expect(Container.get(EngineConfig).mode).toBe('in-process');
expect(consoleWarnMock).toHaveBeenCalledWith(expect.stringContaining('N8N_ENGINE_MODE'));
});
});

View File

@ -1,7 +1,7 @@
export const INSTANCE_ID_HEADER = 'n8n-instance-id';
export const INSTANCE_VERSION_HEADER = 'n8n-version';
export const INSTANCE_TYPES = ['main', 'webhook', 'worker'] as const;
export const INSTANCE_TYPES = ['main', 'webhook', 'worker', 'engine'] as const;
export type InstanceType = (typeof INSTANCE_TYPES)[number];
export const INSTANCE_ROLES = ['unset', 'leader', 'follower'] as const;

View File

@ -0,0 +1,199 @@
import { mockInstance } from '@n8n/backend-test-utils';
import { EngineConfig } from '@n8n/config';
import { DbConnection } from '@n8n/db';
import { ErrorReporter } from 'n8n-core';
import * as CrashJournal from '@/crash-journal';
import { EncryptionBootstrapService } from '@/encryption/encryption-bootstrap.service';
import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
import { ActivityEventRelay } from '@/events/relays/activity.event-relay';
import { TelemetryEventRelay } from '@/events/relays/telemetry.event-relay';
import { WorkflowFailureNotificationEventRelay } from '@/events/relays/workflow-failure-notification.event-relay';
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
import { EngineV2Runtime } from '@/modules/engine-v2/engine-v2.runtime';
import { NodeTypes } from '@/node-types';
import { OtelService } from '@/modules/otel/otel.service';
import { PostHogClient } from '@/posthog';
import { ShutdownService } from '@/shutdown/shutdown.service';
import { TaskRunnerModule } from '@/task-runners/task-runner-module';
import { Engine } from '../engine';
vi.mock('@/crash-journal');
const dbConnection = mockInstance(DbConnection);
const encryptionBootstrap = mockInstance(EncryptionBootstrapService);
const loadNodesAndCredentials = mockInstance(LoadNodesAndCredentials);
loadNodesAndCredentials.init.mockResolvedValue(undefined);
loadNodesAndCredentials.postProcessLoaders.mockResolvedValue(undefined);
const runtime = mockInstance(EngineV2Runtime);
const taskRunnerModule = mockInstance(TaskRunnerModule);
// Services the base `init()` reaches, as in worker.test.ts.
const errorReporter = mockInstance(ErrorReporter);
mockInstance(NodeTypes);
mockInstance(ShutdownService);
mockInstance(MessageEventBus);
mockInstance(PostHogClient);
mockInstance(OtelService);
mockInstance(TelemetryEventRelay);
mockInstance(ActivityEventRelay);
mockInstance(WorkflowFailureNotificationEventRelay);
describe('Engine', () => {
const originalEnv = process.env;
let engineConfig: EngineConfig;
beforeEach(() => {
vi.clearAllMocks();
// The package test script sets `DB_TYPE`, which the command refuses. Start
// from an env with no control plane database and no encryption key.
process.env = Object.fromEntries(
Object.entries(originalEnv).filter(
([key]) =>
!key.startsWith('DB_') &&
key !== 'N8N_ENCRYPTION_KEY' &&
key !== 'N8N_ENCRYPTION_KEY_FILE',
),
);
engineConfig = mockInstance(EngineConfig, {
authSecret: 'a'.repeat(32),
controlPlaneBaseUrl: 'http://n8n-main:3001',
});
});
afterEach(() => {
process.env = originalEnv;
});
const createEngine = () => {
const engine = new Engine();
// @ts-expect-error - Overriding readonly property for testing
engine.globalConfig = {
executions: { mode: 'regular' },
multiMainSetup: { enabled: false },
endpoints: {
metrics: { enable: false },
health: '/health',
webhook: 'webhook',
rest: 'rest',
},
database: { type: 'sqlite' },
sentry: { backendDsn: '' },
cache: { backend: 'memory' },
taskRunners: {},
outboundProxy: { mode: 'main-only' },
expressionEngine: { engine: 'legacy', poolSize: 1, maxCodeCacheSize: 1024 },
generic: { gracefulShutdownTimeout: 30 },
};
return engine;
};
describe('init', () => {
it('refuses to boot with control plane database access', async () => {
process.env.DB_POSTGRESDB_HOST = 'postgres';
await expect(createEngine().init()).rejects.toThrow('DB_POSTGRESDB_HOST');
expect(loadNodesAndCredentials.init).not.toHaveBeenCalled();
});
it('refuses to boot with the control plane encryption key', async () => {
process.env.N8N_ENCRYPTION_KEY = 'secret';
await expect(createEngine().init()).rejects.toThrow('N8N_ENCRYPTION_KEY');
expect(loadNodesAndCredentials.init).not.toHaveBeenCalled();
});
it('refuses to boot with a control plane encryption key file', async () => {
process.env.N8N_ENCRYPTION_KEY_FILE = '/tmp/key';
await expect(createEngine().init()).rejects.toThrow('N8N_ENCRYPTION_KEY_FILE');
expect(loadNodesAndCredentials.init).not.toHaveBeenCalled();
});
it('reports a refusal through the crash path, which the base init has not wired yet', async () => {
process.env.DB_POSTGRESDB_HOST = 'postgres';
const exit = vi.spyOn(process, 'exit').mockImplementation(() => undefined as never);
vi.useFakeTimers();
try {
const engine = createEngine();
const error = await engine.init().catch((e: unknown) => e);
const caught = engine.catch(error as Error);
await vi.runAllTimersAsync();
await caught;
expect(errorReporter.error).toHaveBeenCalledWith(expect.any(Error), { level: 'fatal' });
expect(exit).toHaveBeenCalledWith(1);
} finally {
vi.useRealTimers();
exit.mockRestore();
}
});
it('refuses to boot without the shared secret', async () => {
engineConfig.authSecret = '';
await expect(createEngine().init()).rejects.toThrow('N8N_ENGINE_AUTH_SECRET');
});
it('refuses to boot without the control plane address', async () => {
engineConfig.controlPlaneBaseUrl = '';
await expect(createEngine().init()).rejects.toThrow('N8N_ENGINE_CONTROL_PLANE_BASE_URL');
});
it('never opens, migrates or reads the control plane database', async () => {
await createEngine().init();
expect(dbConnection.init).not.toHaveBeenCalled();
expect(dbConnection.migrate).not.toHaveBeenCalled();
expect(encryptionBootstrap.run).not.toHaveBeenCalled();
});
it('starts no crash journal', async () => {
await createEngine().init();
expect(CrashJournal.init).not.toHaveBeenCalled();
});
it('loads the nodes and starts no task runner', async () => {
await createEngine().init();
expect(loadNodesAndCredentials.init).toHaveBeenCalled();
expect(taskRunnerModule.start).not.toHaveBeenCalled();
});
it('starts the data plane after the base init and finishes loading the nodes', async () => {
await createEngine().init();
expect(runtime.init).toHaveBeenCalled();
expect(vi.mocked(runtime.init).mock.invocationCallOrder[0]).toBeGreaterThan(
vi.mocked(loadNodesAndCredentials.init).mock.invocationCallOrder[0],
);
expect(loadNodesAndCredentials.postProcessLoaders).toHaveBeenCalled();
});
});
describe('stopProcess', () => {
it('shuts the data plane down before exiting, without touching the crash journal', async () => {
const engine = createEngine();
await engine.init();
const exit = vi.spyOn(process, 'exit').mockImplementation(() => undefined as never);
try {
// @ts-expect-error - Accessing protected method for testing
await engine.stopProcess();
expect(runtime.shutdown).toHaveBeenCalled();
expect(exit).toHaveBeenCalledWith();
expect(vi.mocked(runtime.shutdown).mock.invocationCallOrder[0]).toBeLessThan(
exit.mock.invocationCallOrder[0],
);
expect(CrashJournal.cleanup).not.toHaveBeenCalled();
expect(dbConnection.close).not.toHaveBeenCalled();
} finally {
exit.mockRestore();
}
});
});
});

View File

@ -91,6 +91,13 @@ export abstract class BaseCommand<F = never> {
/** Whether to init community packages (if enabled) */
protected needsCommunityPackages = false;
/**
* Whether to connect to the control plane database. The engine data plane
* process runs without one, so it also skips the migrations, the persisted
* instance identity and the encryption bootstrap that need it.
*/
protected needsDb = true;
/** Whether to init task runner. */
protected needsTaskRunner = false;
@ -169,47 +176,49 @@ export abstract class BaseCommand<F = never> {
Container.get(LockService).setProvider(Container.get(RedisLockService));
}
await this.dbConnection
.init()
.catch(
async (error: Error) =>
await this.exitWithCrash('There was an error initializing DB', error),
);
if (this.needsDb) {
await this.dbConnection
.init()
.catch(
async (error: Error) =>
await this.exitWithCrash('There was an error initializing DB', error),
);
// This needs to happen after DB.init() or otherwise DB Connection is not
// available via the dependency Container that services depend on.
if (inDevelopment || inTest) {
this.shutdownService.validate();
// This needs to happen after DB.init() or otherwise DB Connection is not
// available via the dependency Container that services depend on.
if (inDevelopment || inTest) {
this.shutdownService.validate();
}
await this.server?.init();
await this.dbConnection
.migrate()
.catch(
async (error: Error) =>
await this.exitWithCrash('There was an error running database migrations', error),
);
// Apply the persisted instance identity so every command (e.g. license:info)
// sees the same instanceId as the running server. Non-fatal for one-off
// commands, which must keep working with restricted DB credentials.
try {
await this.instanceSettings.initialize(Container.get(DeploymentKeyRepository), {
canSeed: this.seedsInstanceIdentity,
});
} catch (error) {
if (this.seedsInstanceIdentity) throw error;
this.logger.warn('Could not read the instance identity from the DB, using derived values', {
error: ensureError(error),
});
}
// Wire the encryption key provider (and seed keys on a seeding main) before
// anything encrypts or decrypts. This must run for every entrypoint —
// servers and one-off commands — since the cipher has no fallback path.
await Container.get(EncryptionBootstrapService).run();
}
await this.server?.init();
await this.dbConnection
.migrate()
.catch(
async (error: Error) =>
await this.exitWithCrash('There was an error running database migrations', error),
);
// Apply the persisted instance identity so every command (e.g. license:info)
// sees the same instanceId as the running server. Non-fatal for one-off
// commands, which must keep working with restricted DB credentials.
try {
await this.instanceSettings.initialize(Container.get(DeploymentKeyRepository), {
canSeed: this.seedsInstanceIdentity,
});
} catch (error) {
if (this.seedsInstanceIdentity) throw error;
this.logger.warn('Could not read the instance identity from the DB, using derived values', {
error: ensureError(error),
});
}
// Wire the encryption key provider (and seed keys on a seeding main) before
// anything encrypts or decrypts. This must run for every entrypoint —
// servers and one-off commands — since the cipher has no fallback path.
await Container.get(EncryptionBootstrapService).run();
if (process.env.EXECUTIONS_PROCESS === 'own') process.exit(-1);
if (

View File

@ -0,0 +1,106 @@
import { EngineConfig } from '@n8n/config';
import { Command } from '@n8n/decorators';
import { Container } from '@n8n/di';
import { ErrorReporter } from 'n8n-core';
import { Expression, UserError } from 'n8n-workflow';
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
import type { EngineV2Runtime } from '@/modules/engine-v2/engine-v2.runtime';
import { BaseCommand } from './base-command';
@Command({
name: 'engine',
description:
'Starts the engine 2.0 data plane. Needs a control plane (`n8n start` with N8N_ENGINE_MODE=remote) to report to and to resolve credentials from.',
})
export class Engine extends BaseCommand {
// The data plane has no control plane database. Its own database is the
// engine's, opened by the runtime from N8N_ENGINE_DATABASE_URL.
override needsDb = false;
// No task runner yet, so the Code node fails with a clear error instead of
// waiting for a runner that never connects.
override needsTaskRunner = false;
override needsExpressionEngine = true;
private runtime?: EngineV2Runtime;
async init() {
// The guards below run before `super.init()` wires the reporter, and the
// crash path reports through it.
this.errorReporter = Container.get(ErrorReporter);
assertControlPlaneIsolated(process.env);
assertRemoteControlPlane(Container.get(EngineConfig));
this.logger.info('Starting engine 2.0 data plane...');
this.logger.debug(`Host ID: ${this.instanceSettings.hostId}`);
await super.init();
const { EngineV2Runtime } = await import('@/modules/engine-v2/engine-v2.runtime.js');
this.runtime = Container.get(EngineV2Runtime);
await this.runtime.init();
await Container.get(LoadNodesAndCredentials).postProcessLoaders();
}
async run() {
this.logger.info('Engine 2.0 data plane waiting for executions.');
// Make sure that the process does not close
await new Promise(() => {});
}
async catch(error: Error) {
await this.exitWithCrash('Exiting due to an error.', error);
}
protected async stopProcess() {
this.logger.info('Stopping engine 2.0 data plane...');
try {
await this.runtime?.shutdown();
await Expression.disposeExpressionEngine();
} catch (error) {
await this.exitWithCrash('There was an error shutting down the engine.', error);
}
// No control plane database to close and no crash journal to clear, so
// the base exit helper does not apply.
process.exit();
}
}
/**
* A `DB_*` variable means the process was given the control plane database,
* and `N8N_ENCRYPTION_KEY` means it could read the credentials in it. Refuse
* both here, at boot, so the isolation is a check and not a convention.
*/
function assertControlPlaneIsolated(env: NodeJS.ProcessEnv): void {
const leaked = Object.keys(env).filter(
(key) =>
key.startsWith('DB_') || key === 'N8N_ENCRYPTION_KEY' || key === 'N8N_ENCRYPTION_KEY_FILE',
);
if (leaked.length > 0) {
throw new UserError(
`The engine process must not have control plane database access. Remove ${leaked.join(', ')} from its environment.`,
);
}
}
function assertRemoteControlPlane(config: EngineConfig): void {
if (!config.authSecret) {
throw new UserError(
'The engine process needs N8N_ENGINE_AUTH_SECRET. The control plane runs elsewhere, so nothing here can generate the shared secret.',
);
}
if (!config.controlPlaneBaseUrl) {
throw new UserError(
'The engine process needs N8N_ENGINE_CONTROL_PLANE_BASE_URL. The default points at this process, which does not run the control plane.',
);
}
}

View File

@ -22,7 +22,7 @@ describe('EngineV2Module', () => {
vi.clearAllMocks();
executionsConfig = mockInstance(ExecutionsConfig, { mode: 'regular' });
engineConfig = mockInstance(EngineConfig, { authSecret: '' });
engineConfig = mockInstance(EngineConfig, { authSecret: '', mode: 'in-process' });
runtime = mockInstance(EngineV2Runtime);
client = mockInstance(EngineDataPlaneClient);
controlPlaneServer = mockInstance(EngineControlPlaneServer);
@ -87,6 +87,67 @@ describe('EngineV2Module', () => {
});
});
describe('remote mode', () => {
beforeEach(() => {
engineConfig.mode = 'remote';
engineConfig.authSecret = 'a'.repeat(32);
engineConfig.baseUrl = 'http://engine:3000';
});
it('starts the control plane server and not the data plane', async () => {
await module.init();
expect(controlPlaneServer.start).toHaveBeenCalled();
expect(runtime.init).not.toHaveBeenCalled();
});
it('registers the client as the data plane provider', async () => {
await module.init();
const proxy = Container.get(EngineDataPlaneProxyService);
await proxy.startExecution({
workflowId: 'wf-1',
graph: { nodes: [], edges: [] },
workflow: {},
executionId: '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa',
callerContext: { hostMode: 'trigger' },
});
expect(client.startExecution).toHaveBeenCalled();
});
it('refuses to start without a shared secret', async () => {
engineConfig.authSecret = '';
await expect(module.init()).rejects.toThrow('N8N_ENGINE_AUTH_SECRET');
expect(controlPlaneServer.start).not.toHaveBeenCalled();
});
it('does not generate a secret, since the data plane cannot learn it', async () => {
engineConfig.authSecret = '';
await module.init().catch(() => {});
expect(engineConfig.authSecret).toBe('');
});
it('refuses to start without the data plane address', async () => {
engineConfig.baseUrl = '';
await expect(module.init()).rejects.toThrow('N8N_ENGINE_BASE_URL');
expect(controlPlaneServer.start).not.toHaveBeenCalled();
});
it('stops only the control plane server on shutdown', async () => {
await module.init();
await module.shutdown();
expect(controlPlaneServer.stop).toHaveBeenCalled();
expect(runtime.shutdown).not.toHaveBeenCalled();
});
});
describe('shutdown', () => {
it('shuts the runtime down', async () => {
await module.shutdown();

View File

@ -8,14 +8,17 @@ import { UserError } from 'n8n-workflow';
import { randomBytes } from 'node:crypto';
/**
* Runs the engine 2.0 data plane in-process.
* Runs the control plane side of engine 2.0, and in `in-process` mode the data
* plane too.
*
* Not a default module: enable it with `N8N_ENABLED_MODULES=engine-v2`. When it
* is off, nothing here loads and no data plane connection is opened.
*
* Main-only, and regular mode only. The in-process engine uses
* `InMemoryWorkQueue`, so its work does not survive the process and cannot be
* shared with other mains or workers.
* shared with other mains or workers. In `remote` mode (`N8N_ENGINE_MODE`) a
* separate `n8n engine` process hosts the data plane; this module then starts
* only the control plane server and the client that dials the engine.
*/
@BackendModule({ name: 'engine-v2', instanceTypes: ['main'] })
export class EngineV2Module implements ModuleInterface {
@ -29,9 +32,12 @@ export class EngineV2Module implements ModuleInterface {
}
const engineConfig = Container.get(EngineConfig);
// Both planes live in this process, so a generated secret is enough and the
// integrated engine is never unauthenticated. A separate CP must set it.
if (!engineConfig.authSecret) {
if (engineConfig.mode === 'remote') {
assertRemoteConfig(engineConfig);
} else if (!engineConfig.authSecret) {
// Both planes live in this process, so a generated secret is enough and the
// integrated engine is never unauthenticated. A separate DP must be given one.
engineConfig.authSecret = randomBytes(32).toString('hex');
}
@ -39,32 +45,34 @@ export class EngineV2Module implements ModuleInterface {
const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js');
await Container.get(EngineControlPlaneServer).start();
// Hand both endpoints over before the engine starts. A short run can answer
// before `startExecution` returns, and responses are not replayed.
const { InMemoryExecutionResponseChannel } = await import(
'./response-channel/in-memory-execution-response-channel.js'
);
const { InMemoryExecutionResponseSender } = await import(
'./response-channel/in-memory-execution-response-sender.js'
);
const { InMemoryExecutionResponseReceiver } = await import(
'./response-channel/in-memory-execution-response-receiver.js'
);
const { EngineV2WebhookResponder } = await import(
'@/services/engine-v2-webhook-responder.service.js'
);
// In-memory for now because both planes share this process. Redis endpoints
// can use the same response contracts when the planes run separately.
const responseChannel = new InMemoryExecutionResponseChannel();
const scopedLogger = Container.get(Logger).scoped('engine-v2');
const responseSender = new InMemoryExecutionResponseSender(responseChannel, scopedLogger);
const responseReceiver = new InMemoryExecutionResponseReceiver(responseChannel, scopedLogger);
Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver);
this.responseSender = responseSender;
this.responseReceiver = responseReceiver;
if (engineConfig.mode === 'in-process') {
// Hand both endpoints over before the engine starts. A short run can answer
// before `startExecution` returns, and responses are not replayed.
const { InMemoryExecutionResponseChannel } = await import(
'./response-channel/in-memory-execution-response-channel.js'
);
const { InMemoryExecutionResponseSender } = await import(
'./response-channel/in-memory-execution-response-sender.js'
);
const { InMemoryExecutionResponseReceiver } = await import(
'./response-channel/in-memory-execution-response-receiver.js'
);
const { EngineV2WebhookResponder } = await import(
'@/services/engine-v2-webhook-responder.service.js'
);
// In-memory for now because both planes share this process. Redis endpoints
// can use the same response contracts when the planes run separately.
const responseChannel = new InMemoryExecutionResponseChannel();
const scopedLogger = Container.get(Logger).scoped('engine-v2');
const responseSender = new InMemoryExecutionResponseSender(responseChannel, scopedLogger);
const responseReceiver = new InMemoryExecutionResponseReceiver(responseChannel, scopedLogger);
Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver);
this.responseSender = responseSender;
this.responseReceiver = responseReceiver;
const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
await Container.get(EngineV2Runtime).init(responseSender);
const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
await Container.get(EngineV2Runtime).init(responseSender);
}
const { EngineDataPlaneClient } = await import('./engine-data-plane-client.js');
const { EngineDataPlaneProxyService } = await import(
@ -77,13 +85,13 @@ export class EngineV2Module implements ModuleInterface {
@OnShutdown()
async shutdown() {
const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
try {
await Container.get(EngineV2Runtime).shutdown();
} finally {
// After the engine, so a final response still has somewhere to go.
// A failed runtime shutdown must still release both endpoints.
await Promise.all([this.responseSender?.stop(), this.responseReceiver?.stop()]);
if (Container.get(EngineConfig).mode === 'in-process') {
const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
try {
await Container.get(EngineV2Runtime).shutdown();
} finally {
await Promise.all([this.responseSender?.stop(), this.responseReceiver?.stop()]);
}
}
// After the engine, so its final flush still has somewhere to land.
@ -91,3 +99,18 @@ export class EngineV2Module implements ModuleInterface {
await Container.get(EngineControlPlaneServer).stop();
}
}
/** A remote data plane needs an address to dial and a secret it was also given. */
function assertRemoteConfig(config: EngineConfig): void {
if (!config.authSecret) {
throw new UserError(
'N8N_ENGINE_MODE=remote needs N8N_ENGINE_AUTH_SECRET. The data plane runs elsewhere, so this main cannot generate the shared secret.',
);
}
if (!config.baseUrl) {
throw new UserError(
'N8N_ENGINE_MODE=remote needs N8N_ENGINE_BASE_URL. The default points at this process, which does not run the data plane.',
);
}
}

View File

@ -6,6 +6,7 @@ import {
AllowAllAdmittance,
createDataSource,
createEngineRuntime,
noopExecutionResponseSender,
SharedSecretIdentityVerifier,
} from '@n8n/engine';
import type { AdditionalDataContext } from '@n8n/node-engine-compatibility';
@ -60,7 +61,7 @@ export class EngineV2Runtime {
this.logger = this.logger.scoped('engine-v2');
}
async init(responseSender: ExecutionResponseSender): Promise<void> {
async init(responseSender: ExecutionResponseSender = noopExecutionResponseSender): Promise<void> {
try {
await this.initDb();

View File

@ -1,5 +1,7 @@
import type { StepSlots } from '@n8n/engine';
import { WorkflowOperationError } from 'n8n-workflow';
import type { EngineConfig } from '@n8n/config';
import type { INode, WebhookResponseMode } from 'n8n-workflow';
import { UserError, WorkflowOperationError } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
@ -11,8 +13,48 @@ const engineV2Webhooks = new EngineV2Webhooks(
mock<EngineV2Dispatcher>(),
mock<EngineV2PayloadGuard>(),
mock<EngineDataPlaneProxyService>(),
mock<EngineConfig>(),
);
describe('EngineV2Webhooks.assertSupported', () => {
const assertSupported = (mode: EngineConfig['mode'], responseMode: WebhookResponseMode) => {
const proxy = mock<EngineDataPlaneProxyService>();
proxy.isAvailable.mockReturnValue(true);
const webhooks = new EngineV2Webhooks(
mock<EngineV2Dispatcher>(),
mock<EngineV2PayloadGuard>(),
proxy,
mock<EngineConfig>({ mode }),
);
webhooks.assertSupported({
workflowStartNode: mock<INode>({
name: 'Webhook',
type: 'n8n-nodes-base.webhook',
parameters: {},
}),
responseMode,
executionId: undefined,
});
};
it.each(['lastNode', 'responseNode'] as const)(
'rejects %s responses with a remote data plane',
(responseMode) => {
expect(() => assertSupported('remote', responseMode)).toThrow(UserError);
expect(() => assertSupported('remote', responseMode)).toThrow('remote data plane');
},
);
it('allows onReceived with a remote data plane', () => {
expect(() => assertSupported('remote', 'onReceived')).not.toThrow();
});
it.each(['lastNode', 'responseNode'] as const)('allows %s in-process', (responseMode) => {
expect(() => assertSupported('in-process', responseMode)).not.toThrow();
});
});
describe('EngineV2Webhooks.toRun', () => {
it('converts a completed outcome with outputs', async () => {
const outputs: StepSlots = [[{ json: { answer: 42 } }], null];

View File

@ -1,3 +1,4 @@
import { EngineConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import type {
INode,
@ -72,6 +73,7 @@ export class EngineV2Webhooks {
private readonly dispatcher: EngineV2Dispatcher,
private readonly payloadGuard: EngineV2PayloadGuard,
private readonly proxy: EngineDataPlaneProxyService,
private readonly engineConfig: EngineConfig,
) {}
/** Whether this webhook run starts on the engine 2.0 data plane. */
@ -131,6 +133,11 @@ export class EngineV2Webhooks {
`Engine 2.0 does not support the '${responseMode}' response mode yet. Respond immediately instead.`,
);
}
if (this.engineConfig.mode === 'remote' && responseMode !== 'onReceived') {
throw new UserError(
`Engine 2.0 does not support the '${responseMode}' response mode with a remote data plane yet. Respond immediately instead.`,
);
}
}
/** Converts the data plane's answer to the shape the v1 response path reads. */

View File

@ -595,6 +595,19 @@ describe('NodeExecutionContext', () => {
);
nodeTypes.getByNameAndVersion.mockReturnValue(nodeType);
});
it('rejects signed resume URLs in the engine process', () => {
const previousInstanceType = instanceSettings.instanceType;
Object.assign(instanceSettings, { instanceType: 'engine' });
try {
expect(() => testContext.getSignedResumeUrl()).toThrow(
'Engine v2 does not support signed resume URLs yet',
);
} finally {
Object.assign(instanceSettings, { instanceType: previousInstanceType });
}
});
it('should return a resume URL with HMAC signature', () => {
const result = testContext.getSignedResumeUrl();

View File

@ -272,6 +272,10 @@ export abstract class NodeExecutionContext implements Omit<FunctionsBase, 'getCr
}
getSignedResumeUrl(parameters: Record<string, string> = {}) {
if (this.instanceSettings.instanceType === 'engine') {
throw new UserError('Engine v2 does not support signed resume URLs yet');
}
const { webhookWaitingBaseUrl, executionId } = this.additionalData;
if (typeof executionId !== 'string') {

View File

@ -208,6 +208,17 @@ describe('InstanceSettings', () => {
expect(instanceType).toEqual('main');
expect(hostId.length).toBeGreaterThan(0); // hostname or nanoID
});
it('should detect the engine command as its own instance type', () => {
process.argv[2] = 'engine';
mockFs.existsSync.mockReturnValue(true);
mockFs.readFileSync.mockReturnValue(JSON.stringify({ encryptionKey: 'test_key' }));
const settings = createInstanceSettings();
expect(settings.instanceType).toBe('engine');
expect(settings.hostId.startsWith('engine-')).toBe(true);
});
});
describe('nodeDefinitionsDir', () => {

View File

@ -98,7 +98,7 @@ export class InstanceSettings {
private readonly logger: Logger,
) {
const command = process.argv[2] as InstanceType;
this.instanceType = ['webhook', 'worker'].includes(command) ? command : 'main';
this.instanceType = ['webhook', 'worker', 'engine'].includes(command) ? command : 'main';
this.hostId = `${this.instanceType}-${this.isDocker ? os.hostname() : nanoid()}`;
this.settings = this.loadOrCreate();

View File

@ -2,15 +2,13 @@ import { useRootStore } from '@n8n/stores/useRootStore';
import { useSettingsStore } from '@n8n/stores/settings.store';
import { useInstanceRegistryStore } from '@n8n/frontend-module-instance-registry';
import { useDeviceSupport } from '@n8n/composables/useDeviceSupport';
import type { InstanceRegistration } from '@n8n/api-types';
import type { WorkflowSettings } from 'n8n-workflow';
type ClusterInstanceSummary = {
instanceKey: string;
hostId: string;
instanceType: 'main' | 'worker' | 'webhook';
instanceRole: 'leader' | 'follower' | 'unset';
version: string;
};
type ClusterInstanceSummary = Pick<
InstanceRegistration,
'instanceKey' | 'hostId' | 'instanceType' | 'instanceRole' | 'version'
>;
type ClusterCheckSummary = {
check: string;