refactor(engine): Extract the engine runtime wiring into a factory (no-changelog) (#36504)

packages/@n8n/engine/AGENTS.md

@@ -35,9 +35,18 @@ a deployable engine worker) without touching core logic.
   (in-memory default). The Postgres/ORM coupling lives *here only*.
 - **Serving infra** — `server/` (express), `serve.ts` (standalone entrypoint),
   Dockerfile. First candidate to be extracted later.
+- **Runtime factory** — `runtime/` (`createEngineRuntime`). Owns the engine's
+  internal topology: the queues, the stores, the handlers, the workers, the HTTP
+  app and the start/stop order. It takes the adapters a host chooses and returns
+  a running engine, so no host repeats the wiring. A data plane database is not
+  optional — the factory's type requires one, and each composition root refuses
+  to start without it rather than serving `/healthz` while unable to run a
+  workflow.
 - **Composition roots** — `serve.ts` (standalone) and, in integrated mode,
-  `packages/cli`. These construct the concrete adapters (the `DataSource`, etc.)
-  and hand them in. **Construction lives here, not in the core.**
+  `packages/cli`. These construct the concrete adapters (the `DataSource`, the
+  admittance policy, the v1 step executor) and hand them to
+  `createEngineRuntime`. **Construction lives here, not in the core** — but the
+  topology does not: that belongs to the factory.
 
 ## Rules that keep the seams extractable
 

packages/@n8n/engine/src/database/create-stores.ts

@@ -6,10 +6,13 @@ import { TypeOrmStepStore } from './typeorm-step-store';
 import type { ExecutionStore } from '../execution/execution-store';
 import type { StepStore } from '../execution/step-store';
 
-export function createStores(dataSource: DataSource): {
+/** The engine's own persistence, for hosts that must read the data it writes. */
+export interface EngineStores {
 	executionStore: ExecutionStore;
 	stepStore: StepStore;
-} {
+}
+
+export function createStores(dataSource: DataSource): EngineStores {
 	return {
 		executionStore: new TypeOrmExecutionStore(dataSource.getRepository(WorkflowExecution)),
 		stepStore: new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution)),

packages/@n8n/engine/src/database/index.ts

@@ -1,5 +1,6 @@
 export { createDataSource } from './data-source';
 export { createStores } from './create-stores';
+export type { EngineStores } from './create-stores';
 export { WorkflowExecution, WorkflowStepExecution } from './entities';
 export { TypeOrmExecutionStore } from './typeorm-execution-store';
 export { TypeOrmStepStore } from './typeorm-step-store';

packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts

@@ -1,26 +1,23 @@
 import type { DataSource } from '@n8n/typeorm';
 import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql';
 import postgresVersions from 'n8n-containers/postgres-versions.json';
+import request from 'supertest';
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
 
 import { AllowAllAdmittance } from '../../admittance';
 import {
 	createDataSource,
-	TypeOrmExecutionStore,
-	TypeOrmStepStore,
+	createStores,
 	WorkflowExecution,
 	WorkflowStepExecution,
 } from '../../database';
 import type { IStepExecutor, StepExecutionRequest } from '../../dependencies';
 import type { WorkflowGraph } from '../../graph';
-import { InMemoryWorkQueue, type OrchestrationMessage, type StepMessage } from '../../queue';
-import { ExecutionStartHandler } from '../execution-start-handler';
+import { InMemoryWorkQueue, type OrchestrationMessage } from '../../queue';
+import { createEngineRuntime } from '../../runtime';
 import type { TriggerOutputs } from '../execution.types';
-import { OrchestrationWorker } from '../orchestration-worker';
-import { StartExecutionService } from '../start-execution.service';
+import type { StartExecutionResult } from '../start-execution.service';
 import { StepReadyHandler } from '../step-ready-handler';
-import { StepSettledHandler } from '../step-settled-handler';
-import { StepWorker } from '../step-worker';
 
 const graph: WorkflowGraph = {
 	nodes: [
@@ -46,59 +43,43 @@ describe('step execution (integration)', () => {
 		if (container) await container.stop();
 	});
 
-	function stores() {
-		return {
-			executionStore: new TypeOrmExecutionStore(dataSource.getRepository(WorkflowExecution)),
-			stepStore: new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution)),
-		};
-	}
-
 	/**
-	 * Wires both workers over shared queues and runs a workflow through them.
-	 * Resolves once the execution's outcome is recorded, which is after every
-	 * step's own outcome is durable.
+	 * Runs a workflow through a real engine runtime. Resolves once the
+	 * execution's outcome is recorded, which is after every step's own outcome is
+	 * durable.
 	 */
 	async function runWorkflow(
 		executor: IStepExecutor,
 		triggerOutputs: TriggerOutputs,
 		{ workflowId = 'wf-1', graph: workflowGraph = graph } = {},
 	) {
-		const { executionStore, stepStore } = stores();
-		const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
-		const stepQueue = new InMemoryWorkQueue<StepMessage>();
-
 		let done!: () => void;
 		const finished = new Promise<void>((resolve) => (done = resolve));
-		const finishExecution = executionStore.finishExecution.bind(executionStore);
-		vi.spyOn(executionStore, 'finishExecution').mockImplementation(async (id, status) => {
-			const recorded = await finishExecution(id, status);
-			done();
-			return recorded;
+
+		const runtime = createEngineRuntime({
+			dataSource,
+			admittance: new AllowAllAdmittance(),
+			// also how the test reaches the stores the runtime owns
+			externalDependencies: ({ executionStore }) => {
+				const finishExecution = executionStore.finishExecution.bind(executionStore);
+				vi.spyOn(executionStore, 'finishExecution').mockImplementation(async (id, status) => {
+					const recorded = await finishExecution(id, status);
+					done();
+					return recorded;
+				});
+				return { v1StepExecutor: executor };
+			},
 		});
+		runtime.start();
 
-		const orchestrationWorker = new OrchestrationWorker(
-			orchestrationQueue,
-			new ExecutionStartHandler(executionStore, stepStore, orchestrationQueue),
-			new StepSettledHandler(executionStore, stepStore, stepQueue, orchestrationQueue),
-		);
-		const stepWorker = new StepWorker(
-			stepQueue,
-			new StepReadyHandler(executionStore, stepStore, orchestrationQueue, {
-				v1StepExecutor: executor,
-			}),
-		);
-		orchestrationWorker.start();
-		stepWorker.start();
-
-		const { executionId } = await new StartExecutionService(
-			new AllowAllAdmittance(),
-			executionStore,
-			orchestrationQueue,
-		).start({ workflowId, graph: workflowGraph, triggerOutputs });
+		const response = await request(runtime.app)
+			.post('/api/workflow-executions')
+			.send({ workflowId, graph: workflowGraph, triggerOutputs })
+			.expect(201);
+		const { executionId } = response.body as StartExecutionResult;
 		await finished;
 
-		await stepWorker.stop();
-		await orchestrationWorker.stop();
+		await runtime.stop();
 
 		const execution = await dataSource
 			.getRepository(WorkflowExecution)
@@ -304,7 +285,7 @@ describe('step execution (integration)', () => {
 	});
 
 	it('is idempotent across duplicate step:ready deliveries', async () => {
-		const { executionStore, stepStore } = stores();
+		const { executionStore, stepStore } = createStores(dataSource);
 		const execute = vi.fn().mockResolvedValue({ outputs: [[{ json: { n: 1 } }]] });
 		const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
 		const handler = new StepReadyHandler(executionStore, stepStore, orchestrationQueue, {

packages/@n8n/engine/src/index.ts

@@ -1,5 +1,5 @@
-export { createEngineServer } from './server';
-export type { EngineServerDeps } from './server';
+export { createEngineRuntime } from './runtime';
+export type { EngineRuntime, EngineRuntimeOptions } from './runtime';
 
 export type { JsonObject, JsonValue } from './common';
 
@@ -26,7 +26,6 @@ export type {
 	AdmittanceService,
 } from './admittance';
 
-export { InMemoryWorkQueue } from './queue';
 export type {
 	ExecutionEnqueuedEvent,
 	OrchestrationMessage,
@@ -36,16 +35,7 @@ export type {
 	WorkQueue,
 } from './queue';
 
-export {
-	ExecutionNotFoundError,
-	ExecutionStartHandler,
-	OrchestrationWorker,
-	StartExecutionService,
-	StepNotFoundError,
-	StepReadyHandler,
-	StepSettledHandler,
-	StepWorker,
-} from './execution';
+export { ExecutionNotFoundError, StepNotFoundError } from './execution';
 export type {
 	ExecutionMode,
 	ExecutionRecord,
@@ -65,9 +55,5 @@ export type {
 	TriggerOutputs,
 } from './execution';
 
-export {
-	createDataSource,
-	createStores,
-	WorkflowExecution,
-	WorkflowStepExecution,
-} from './database';
+export { createDataSource, WorkflowExecution, WorkflowStepExecution } from './database';
+export type { EngineStores } from './database';

packages/@n8n/engine/src/runtime/__tests__/create-engine-runtime.test.ts

@@ -0,0 +1,55 @@
+import type { DataSource } from '@n8n/typeorm';
+import request from 'supertest';
+import { describe, expect, it, vi } from 'vitest';
+
+import { AllowAllAdmittance } from '../../admittance';
+import type { EngineStores } from '../../database';
+import { createEngineRuntime } from '../create-engine-runtime';
+
+/** Enough of a `DataSource` for the stores: they only hold on to a repository. */
+const fakeDataSource = () => ({ getRepository: vi.fn(() => ({})) }) as unknown as DataSource;
+
+const runtime = () =>
+	createEngineRuntime({
+		dataSource: fakeDataSource(),
+		admittance: new AllowAllAdmittance(),
+	});
+
+describe('createEngineRuntime', () => {
+	it('mounts the execution API', async () => {
+		// a rejected body proves the route is mounted without reaching the store
+		const response = await request(runtime().app).post('/api/workflow-executions').send({});
+
+		expect(response.status).toBe(400);
+	});
+
+	it('serves the healthcheck', async () => {
+		const response = await request(runtime().app).get('/healthz');
+
+		expect(response.status).toBe(200);
+		expect(response.body).toEqual({ status: 'ok' });
+	});
+
+	it('hands the engine stores to the external dependencies', () => {
+		let stores: EngineStores | undefined;
+
+		createEngineRuntime({
+			dataSource: fakeDataSource(),
+			admittance: new AllowAllAdmittance(),
+			externalDependencies: (given) => {
+				stores = given;
+				return {};
+			},
+		});
+
+		expect(stores?.executionStore).toBeDefined();
+		expect(stores?.stepStore).toBeDefined();
+	});
+
+	it('starts and stops both workers', async () => {
+		const engine = runtime();
+		engine.start();
+
+		await expect(engine.stop()).resolves.toBeUndefined();
+	});
+});

packages/@n8n/engine/src/runtime/create-engine-runtime.ts

@@ -0,0 +1,97 @@
+import type { DataSource } from '@n8n/typeorm';
+import type { Application } from 'express';
+
+import type { AdmittanceService } from '../admittance';
+import { createStores } from '../database';
+import type { EngineStores } from '../database';
+import type { ExternalDependencies } from '../dependencies';
+import {
+	ExecutionStartHandler,
+	OrchestrationWorker,
+	StartExecutionService,
+	StepReadyHandler,
+	StepSettledHandler,
+	StepWorker,
+} from '../execution';
+import { InMemoryWorkQueue } from '../queue';
+import type { OrchestrationMessage, StepMessage } from '../queue';
+import { createEngineServer } from '../server';
+
+export interface EngineRuntimeOptions {
+	/** The data plane database, already initialized and migrated. */
+	dataSource: DataSource;
+	admittance: AdmittanceService;
+	/**
+	 * Builds the capabilities the engine does not own. It receives the engine's
+	 * stores, because a `v1-node` executor reads step data through them and the
+	 * runtime owns them.
+	 *
+	 * Standalone mode omits it, so `v1-node` steps fail as unimplemented: the v1
+	 * executor lives in `@n8n/node-engine-compatibility`, which depends on this
+	 * package, so only an integrated host can supply it.
+	 */
+	externalDependencies?: (stores: EngineStores) => ExternalDependencies;
+}
+
+/** A built engine, ready for a host to serve. */
+export interface EngineRuntime {
+	/** The engine HTTP app. The host decides where, and whether, to listen. */
+	app: Application;
+	/** Starts consuming the engine's queues. */
+	start(): void;
+	/** Stops the engine's workers. The host still owns its listener and its `DataSource`. */
+	stop(): Promise<void>;
+}
+
+/**
+ * Builds a working engine from the adapters a host chooses.
+ *
+ * The engine keeps its internal topology here — which handler goes into which
+ * worker, which queue feeds which handler, and the start and stop order — so no
+ * host can get it wrong. A host keeps only its own choices: the database, the
+ * admittance policy, the external dependencies and the HTTP listener.
+ */
+export function createEngineRuntime({
+	dataSource,
+	admittance,
+	externalDependencies,
+}: EngineRuntimeOptions): EngineRuntime {
+	const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
+	const stepQueue = new InMemoryWorkQueue<StepMessage>();
+	const { executionStore, stepStore } = createStores(dataSource);
+
+	const orchestrationWorker = new OrchestrationWorker(
+		orchestrationQueue,
+		new ExecutionStartHandler(executionStore, stepStore, orchestrationQueue),
+		new StepSettledHandler(executionStore, stepStore, stepQueue, orchestrationQueue),
+	);
+	const stepWorker = new StepWorker(
+		stepQueue,
+		new StepReadyHandler(
+			executionStore,
+			stepStore,
+			orchestrationQueue,
+			externalDependencies?.({ executionStore, stepStore }) ?? {},
+		),
+	);
+
+	const { app } = createEngineServer(
+		new StartExecutionService(admittance, executionStore, orchestrationQueue),
+	);
+
+	return {
+		app,
+
+		start: () => {
+			orchestrationWorker.start();
+			stepWorker.start();
+		},
+
+		stop: async () => {
+			// TODO(CAT-3882): drain in-flight work instead. Stopping a worker waits
+			// only for whatever it is mid-handling; anything queued behind it is
+			// dropped, since the in-memory queues die with the process.
+			await Promise.all([orchestrationWorker.stop(), stepWorker.stop()]);
+		},
+	};
+}

packages/@n8n/engine/src/runtime/index.ts

@@ -0,0 +1,2 @@
+export { createEngineRuntime } from './create-engine-runtime';
+export type { EngineRuntime, EngineRuntimeOptions } from './create-engine-runtime';

packages/@n8n/engine/src/serve.ts

@@ -1,64 +1,27 @@
 import { EngineConfig } from '@n8n/config';
 import { Container } from '@n8n/di';
-import type { DataSource } from '@n8n/typeorm';
 
 import { AllowAllAdmittance } from './admittance';
-import { createDataSource, createStores } from './database';
-import {
-	ExecutionStartHandler,
-	OrchestrationWorker,
-	StepSettledHandler,
-	StepReadyHandler,
-	StepWorker,
-} from './execution';
-import { InMemoryWorkQueue } from './queue';
-import type { OrchestrationMessage, StepMessage } from './queue';
-import { createEngineServer } from './server';
+import { createDataSource } from './database';
+import { createEngineRuntime } from './runtime';
 
 async function main(): Promise<void> {
 	const config = Container.get(EngineConfig);
 
-	let dataSource: DataSource | undefined;
-	if (config.databaseUrl) {
-		dataSource = createDataSource(config.databaseUrl);
-		await dataSource.initialize();
-		await dataSource.runMigrations();
-	} else {
-		console.warn(
-			'engine: N8N_ENGINE_DATABASE_URL not set; running in healthcheck-only mode (workflow execution endpoints disabled)',
-		);
+	// Refused rather than degraded: an engine with nowhere to record an execution
+	// would report healthy while being unable to run one.
+	if (!config.databaseUrl) {
+		throw new Error('engine: N8N_ENGINE_DATABASE_URL is not set');
 	}
 
-	const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
-	const stepQueue = new InMemoryWorkQueue<StepMessage>();
+	const dataSource = createDataSource(config.databaseUrl);
+	await dataSource.initialize();
+	await dataSource.runMigrations();
 
-	let orchestrationWorker: OrchestrationWorker | undefined;
-	let stepWorker: StepWorker | undefined;
-	if (dataSource) {
-		const { executionStore, stepStore } = createStores(dataSource);
-		orchestrationWorker = new OrchestrationWorker(
-			orchestrationQueue,
-			new ExecutionStartHandler(executionStore, stepStore, orchestrationQueue),
-			new StepSettledHandler(executionStore, stepStore, stepQueue, orchestrationQueue),
-		);
-		// No executors here: the v1 one lives in `@n8n/node-engine-compatibility`,
-		// which depends on this package, so only an integrated host can supply it.
-		// `v1-node` steps therefore fail as unimplemented in standalone mode.
-		stepWorker = new StepWorker(
-			stepQueue,
-			new StepReadyHandler(executionStore, stepStore, orchestrationQueue, {}),
-		);
-		orchestrationWorker.start();
-		stepWorker.start();
-	}
+	const runtime = createEngineRuntime({ dataSource, admittance: new AllowAllAdmittance() });
+	runtime.start();
 
-	const { app } = createEngineServer(
-		dataSource
-			? { dataSource, admittance: new AllowAllAdmittance(), workQueue: orchestrationQueue }
-			: undefined,
-	);
-
-	const server = app.listen(config.port, config.host, () => {
+	const server = runtime.app.listen(config.port, config.host, () => {
 		console.log(`engine: listening on http://${config.host}:${config.port}`);
 	});
 
@@ -70,12 +33,8 @@ async function main(): Promise<void> {
 		await new Promise<void>((resolve, reject) => {
 			server.close((error) => (error ? reject(error) : resolve()));
 		});
-		// TODO(CAT-3882): drain in-flight work instead. Stopping the workers waits
-		// only for whatever each is mid-handling; anything queued behind it is
-		// dropped, since the in-memory queues die with the process.
-		if (orchestrationWorker) await orchestrationWorker.stop();
-		if (stepWorker) await stepWorker.stop();
-		if (dataSource?.isInitialized) await dataSource.destroy();
+		await runtime.stop();
+		if (dataSource.isInitialized) await dataSource.destroy();
 		process.exit(0);
 	};
 

packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts

@@ -5,7 +5,8 @@ import request from 'supertest';
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { AllowAllAdmittance } from '../../admittance';
-import { createDataSource, WorkflowExecution } from '../../database';
+import { createDataSource, createStores, WorkflowExecution } from '../../database';
+import { StartExecutionService } from '../../execution';
 import type { WorkflowGraph } from '../../graph';
 import type { OrchestrationMessage, WorkQueue } from '../../queue';
 import { startEngineServer } from '../../testing/start-engine-server';
@@ -31,11 +32,10 @@ describe('POST /api/workflow-executions (integration)', () => {
 
 	beforeEach(async () => {
 		workQueue = { publish: vi.fn(), start: vi.fn(), stop: vi.fn() };
-		({ url, stop } = await startEngineServer({
-			dataSource,
-			admittance: new AllowAllAdmittance(),
-			workQueue,
-		}));
+		const { executionStore } = createStores(dataSource);
+		({ url, stop } = await startEngineServer(
+			new StartExecutionService(new AllowAllAdmittance(), executionStore, workQueue),
+		));
 	});
 
 	afterEach(async () => {

packages/@n8n/engine/src/server/create-engine-server.ts

@@ -1,24 +1,10 @@
-import type { DataSource } from '@n8n/typeorm';
 import express, { type Application } from 'express';
 
-import type { AdmittanceService } from '../admittance';
-import { TypeOrmExecutionStore, WorkflowExecution } from '../database';
-import { StartExecutionService } from '../execution/start-execution.service';
-import type { OrchestrationMessage, WorkQueue } from '../queue';
+import type { StartExecutionService } from '../execution/start-execution.service';
 import { createWorkflowExecutionsRouter } from './routes/workflow-executions';
 
-export interface EngineServerDeps {
-	dataSource: DataSource;
-	admittance: AdmittanceService;
-	workQueue: WorkQueue<OrchestrationMessage>;
-}
-
-/**
- * Builds the engine HTTP app. Without `deps` it serves only `/healthz`; with
- * `deps` it also mounts the execution API. Deps are all-or-nothing so the API
- * can't be half-wired.
- */
-export function createEngineServer(deps?: EngineServerDeps): { app: Application } {
+/** Builds the engine HTTP app: `/healthz` plus the execution API. */
+export function createEngineServer(startExecution: StartExecutionService): { app: Application } {
 	const app = express();
 	app.use(express.json());
 
@@ -26,17 +12,7 @@ export function createEngineServer(deps?: EngineServerDeps): { app: Application
 		res.status(200).json({ status: 'ok' });
 	});
 
-	if (deps) {
-		const executionStore = new TypeOrmExecutionStore(
-			deps.dataSource.getRepository(WorkflowExecution),
-		);
-		const startExecution = new StartExecutionService(
-			deps.admittance,
-			executionStore,
-			deps.workQueue,
-		);
-		app.use('/api/workflow-executions', createWorkflowExecutionsRouter(startExecution));
-	}
+	app.use('/api/workflow-executions', createWorkflowExecutionsRouter(startExecution));
 
 	return { app };
 }

packages/@n8n/engine/src/server/index.ts

@@ -1,2 +1 @@
 export { createEngineServer } from './create-engine-server';
-export type { EngineServerDeps } from './create-engine-server';

packages/@n8n/engine/src/testing/__tests__/start-engine-server.integration.test.ts

@@ -1,6 +1,7 @@
 import request from 'supertest';
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';
 
+import type { StartExecutionService } from '../../execution';
 import { startEngineServer } from '../start-engine-server';
 
 describe('engine HTTP server (e2e)', () => {
@@ -8,7 +9,8 @@ describe('engine HTTP server (e2e)', () => {
 	let stop: () => Promise<void>;
 
 	beforeAll(async () => {
-		({ url, stop } = await startEngineServer());
+		// only /healthz is under test, and the execution route never calls the service
+		({ url, stop } = await startEngineServer({} as StartExecutionService));
 	});
 
 	afterAll(async () => {

packages/@n8n/engine/src/testing/start-engine-server.ts

@@ -1,12 +1,13 @@
 import type { Server } from 'node:http';
 
-import { createEngineServer, type EngineServerDeps } from '../server';
+import type { StartExecutionService } from '../execution';
+import { createEngineServer } from '../server';
 
-export async function startEngineServer(deps?: EngineServerDeps): Promise<{
+export async function startEngineServer(startExecution: StartExecutionService): Promise<{
 	url: string;
 	stop: () => Promise<void>;
 }> {
-	const { app } = createEngineServer(deps);
+	const { app } = createEngineServer(startExecution);
 
 	const server = await new Promise<Server>((resolve, reject) => {
 		const s = app.listen(0, '127.0.0.1', () => resolve(s));

packages/@n8n/node-engine-compatibility/package.json

@@ -30,7 +30,9 @@
     "@n8n/typescript-config": "workspace:*",
     "@n8n/vitest-config": "workspace:*",
     "@testcontainers/postgresql": "catalog:",
+    "@types/supertest": "^6.0.3",
     "n8n-nodes-base": "workspace:*",
+    "supertest": "^7.1.1",
     "testcontainers": "catalog:",
     "typescript": "catalog:typescript",
     "vitest": "catalog:"

packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts

@@ -1,20 +1,12 @@
 import type {
 	createDataSource,
-	OrchestrationMessage,
-	StepMessage,
+	StartExecutionResult,
 	TriggerOutputs,
 	WorkflowGraph,
 } from '@n8n/engine';
 import {
 	AllowAllAdmittance,
-	createStores,
-	ExecutionStartHandler,
-	InMemoryWorkQueue,
-	OrchestrationWorker,
-	StartExecutionService,
-	StepReadyHandler,
-	StepSettledHandler,
-	StepWorker,
+	createEngineRuntime,
 	WorkflowExecution,
 	WorkflowStepExecution,
 } from '@n8n/engine';
@@ -26,6 +18,7 @@ import { NoOp } from 'n8n-nodes-base/nodes/NoOp/NoOp.node';
 import { SplitOut } from 'n8n-nodes-base/nodes/Transform/SplitOut/SplitOut.node';
 import type { IDataObject, INodeType, INodeTypes, IVersionedNodeType } from 'n8n-workflow';
 import { NodeHelpers } from 'n8n-workflow';
+import request from 'supertest';
 import { vi } from 'vitest';
 
 import { createEngineStepDataLoader } from '../engine-step-data-loader';
@@ -97,44 +90,39 @@ type EngineDataSource = ReturnType<typeof createDataSource>;
 export function makeRunWorkflow(getDataSource: () => EngineDataSource) {
 	return async function runWorkflow(graph: WorkflowGraph, triggerOutputs: TriggerOutputs | null) {
 		const dataSource = getDataSource();
-		const { executionStore, stepStore } = createStores(dataSource);
-		const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
-		const stepQueue = new InMemoryWorkQueue<StepMessage>();
 
 		let done!: () => void;
 		const finished = new Promise<void>((resolve) => (done = resolve));
-		const finishExecution = executionStore.finishExecution.bind(executionStore);
-		vi.spyOn(executionStore, 'finishExecution').mockImplementation(async (id, status) => {
-			const recorded = await finishExecution(id, status);
-			done();
-			return recorded;
+
+		const runtime = createEngineRuntime({
+			dataSource,
+			admittance: new AllowAllAdmittance(),
+			// also how the test reaches the stores the runtime owns
+			externalDependencies: ({ executionStore, stepStore }) => {
+				const finishExecution = executionStore.finishExecution.bind(executionStore);
+				vi.spyOn(executionStore, 'finishExecution').mockImplementation(async (id, status) => {
+					const recorded = await finishExecution(id, status);
+					done();
+					return recorded;
+				});
+
+				return {
+					v1StepExecutor: new V1StepExecutor({
+						nodeTypes: realNodeTypes,
+						additionalDataFactory: testAdditionalDataFactory,
+						loadStepData: createEngineStepDataLoader(executionStore, stepStore),
+					}),
+				};
+			},
 		});
+		runtime.start();
 
-		const executor = new V1StepExecutor({
-			nodeTypes: realNodeTypes,
-			additionalDataFactory: testAdditionalDataFactory,
-			loadStepData: createEngineStepDataLoader(executionStore, stepStore),
-		});
-
-		const orchestrationWorker = new OrchestrationWorker(
-			orchestrationQueue,
-			new ExecutionStartHandler(executionStore, stepStore, orchestrationQueue),
-			new StepSettledHandler(executionStore, stepStore, stepQueue, orchestrationQueue),
-		);
-		const stepWorker = new StepWorker(
-			stepQueue,
-			new StepReadyHandler(executionStore, stepStore, orchestrationQueue, {
-				v1StepExecutor: executor,
-			}),
-		);
-		orchestrationWorker.start();
-		stepWorker.start();
-
-		const { executionId } = await new StartExecutionService(
-			new AllowAllAdmittance(),
-			executionStore,
-			orchestrationQueue,
-		).start({ workflowId: 'wf-m1', graph, triggerOutputs });
+		// over HTTP, because that is the engine's only boundary
+		const response = await request(runtime.app)
+			.post('/api/workflow-executions')
+			.send({ workflowId: 'wf-m1', graph, triggerOutputs })
+			.expect(201);
+		const { executionId } = response.body as StartExecutionResult;
 
 		try {
 			await Promise.race([
@@ -150,8 +138,7 @@ export function makeRunWorkflow(getDataSource: () => EngineDataSource) {
 				}),
 			]);
 		} finally {
-			await stepWorker.stop();
-			await orchestrationWorker.stop();
+			await runtime.stop();
 		}
 
 		const steps = await dataSource

pnpm-lock.yaml

@@ -3147,9 +3147,15 @@ importers:
       '@testcontainers/postgresql':
         specifier: 'catalog:'
         version: 11.13.0
+      '@types/supertest':
+        specifier: ^6.0.3
+        version: 6.0.3
       n8n-nodes-base:
         specifier: workspace:*
         version: link:../../nodes-base
+      supertest:
+        specifier: ^7.1.1
+        version: 7.1.1
       testcontainers:
         specifier: 'catalog:'
         version: 11.13.0