diff --git a/.github/workflows/ci-pull-requests.yml b/.github/workflows/ci-pull-requests.yml index 3aa555368b6..dc40d38bb7f 100644 --- a/.github/workflows/ci-pull-requests.yml +++ b/.github/workflows/ci-pull-requests.yml @@ -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/** diff --git a/packages/@n8n/api-types/src/instance-registry-types.ts b/packages/@n8n/api-types/src/instance-registry-types.ts index 9eeab861ddc..28321655379 100644 --- a/packages/@n8n/api-types/src/instance-registry-types.ts +++ b/packages/@n8n/api-types/src/instance-registry-types.ts @@ -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']), }), ), diff --git a/packages/@n8n/config/src/configs/engine.config.ts b/packages/@n8n/config/src/configs/engine.config.ts index 7c32bc2f714..9bbaeee777e 100644 --- a/packages/@n8n/config/src/configs/engine.config.ts +++ b/packages/@n8n/config/src/configs/engine.config.ts @@ -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; + @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; diff --git a/packages/@n8n/config/src/index.ts b/packages/@n8n/config/src/index.ts index a93d4f16d8a..94990cbad48 100644 --- a/packages/@n8n/config/src/index.ts +++ b/packages/@n8n/config/src/index.ts @@ -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'; diff --git a/packages/@n8n/config/test/engine.config.test.ts b/packages/@n8n/config/test/engine.config.test.ts index f389e413efe..bbdf7bb182f 100644 --- a/packages/@n8n/config/test/engine.config.test.ts +++ b/packages/@n8n/config/test/engine.config.test.ts @@ -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')); + }); }); diff --git a/packages/@n8n/constants/src/instance.ts b/packages/@n8n/constants/src/instance.ts index 807d41c46ff..41af6d94d94 100644 --- a/packages/@n8n/constants/src/instance.ts +++ b/packages/@n8n/constants/src/instance.ts @@ -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; diff --git a/packages/cli/src/commands/__tests__/engine.test.ts b/packages/cli/src/commands/__tests__/engine.test.ts new file mode 100644 index 00000000000..be059d3dea0 --- /dev/null +++ b/packages/cli/src/commands/__tests__/engine.test.ts @@ -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(); + } + }); + }); +}); diff --git a/packages/cli/src/commands/base-command.ts b/packages/cli/src/commands/base-command.ts index e14c8f719d4..a3bd0a908eb 100644 --- a/packages/cli/src/commands/base-command.ts +++ b/packages/cli/src/commands/base-command.ts @@ -91,6 +91,13 @@ export abstract class BaseCommand { /** 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 { 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 ( diff --git a/packages/cli/src/commands/engine.ts b/packages/cli/src/commands/engine.ts new file mode 100644 index 00000000000..647036e591d --- /dev/null +++ b/packages/cli/src/commands/engine.ts @@ -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.', + ); + } +} 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 2b8bb67cd6a..5b573502115 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 @@ -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(); 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 d491b0ec046..431e40a832a 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.module.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.module.ts @@ -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.', + ); + } +} 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 9546cb351f6..2554f6b4d88 100644 --- a/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts +++ b/packages/cli/src/modules/engine-v2/engine-v2.runtime.ts @@ -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 { + async init(responseSender: ExecutionResponseSender = noopExecutionResponseSender): Promise { try { await this.initDb(); 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 252be4d7b17..b8015b54dca 100644 --- a/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts +++ b/packages/cli/src/webhooks/__tests__/engine-v2-webhooks.test.ts @@ -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(), mock(), mock(), + mock(), ); +describe('EngineV2Webhooks.assertSupported', () => { + const assertSupported = (mode: EngineConfig['mode'], responseMode: WebhookResponseMode) => { + const proxy = mock(); + proxy.isAvailable.mockReturnValue(true); + const webhooks = new EngineV2Webhooks( + mock(), + mock(), + proxy, + mock({ mode }), + ); + + webhooks.assertSupported({ + workflowStartNode: mock({ + 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]; diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index d406e2f4bac..6bc42bc5565 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -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. */ diff --git a/packages/core/src/execution-engine/node-execution-context/__tests__/node-execution-context.test.ts b/packages/core/src/execution-engine/node-execution-context/__tests__/node-execution-context.test.ts index 0866331b9a7..6a33bb98b0b 100644 --- a/packages/core/src/execution-engine/node-execution-context/__tests__/node-execution-context.test.ts +++ b/packages/core/src/execution-engine/node-execution-context/__tests__/node-execution-context.test.ts @@ -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(); diff --git a/packages/core/src/execution-engine/node-execution-context/node-execution-context.ts b/packages/core/src/execution-engine/node-execution-context/node-execution-context.ts index a000e4a7eb7..bfb981a773d 100644 --- a/packages/core/src/execution-engine/node-execution-context/node-execution-context.ts +++ b/packages/core/src/execution-engine/node-execution-context/node-execution-context.ts @@ -272,6 +272,10 @@ export abstract class NodeExecutionContext implements Omit = {}) { + 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') { diff --git a/packages/core/src/instance-settings/__tests__/instance-settings.test.ts b/packages/core/src/instance-settings/__tests__/instance-settings.test.ts index ad0620884c6..68c7e5d9a23 100644 --- a/packages/core/src/instance-settings/__tests__/instance-settings.test.ts +++ b/packages/core/src/instance-settings/__tests__/instance-settings.test.ts @@ -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', () => { diff --git a/packages/core/src/instance-settings/instance-settings.ts b/packages/core/src/instance-settings/instance-settings.ts index e77c894957f..dca25222176 100644 --- a/packages/core/src/instance-settings/instance-settings.ts +++ b/packages/core/src/instance-settings/instance-settings.ts @@ -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(); diff --git a/packages/frontend/editor-ui/src/app/composables/useDebugInfo.ts b/packages/frontend/editor-ui/src/app/composables/useDebugInfo.ts index 96105947b4a..3907536e0ae 100644 --- a/packages/frontend/editor-ui/src/app/composables/useDebugInfo.ts +++ b/packages/frontend/editor-ui/src/app/composables/useDebugInfo.ts @@ -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;