diff --git a/packages/@n8n/engine/AGENTS.md b/packages/@n8n/engine/AGENTS.md index 67dc0b918e2..5a59c254b6d 100644 --- a/packages/@n8n/engine/AGENTS.md +++ b/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 diff --git a/packages/@n8n/engine/src/database/create-stores.ts b/packages/@n8n/engine/src/database/create-stores.ts index a0bf2d85060..2920ff5bf65 100644 --- a/packages/@n8n/engine/src/database/create-stores.ts +++ b/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)), diff --git a/packages/@n8n/engine/src/database/index.ts b/packages/@n8n/engine/src/database/index.ts index 793c8516c88..a093e60bba9 100644 --- a/packages/@n8n/engine/src/database/index.ts +++ b/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'; diff --git a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts index 3ed5f1e9d55..400c4ba7453 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts +++ b/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(); - const stepQueue = new InMemoryWorkQueue(); - let done!: () => void; const finished = new Promise((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(); const handler = new StepReadyHandler(executionStore, stepStore, orchestrationQueue, { diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index 298ca2021c7..b373db0b53e 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/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'; diff --git a/packages/@n8n/engine/src/runtime/__tests__/create-engine-runtime.test.ts b/packages/@n8n/engine/src/runtime/__tests__/create-engine-runtime.test.ts new file mode 100644 index 00000000000..e9f16e43928 --- /dev/null +++ b/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(); + }); +}); diff --git a/packages/@n8n/engine/src/runtime/create-engine-runtime.ts b/packages/@n8n/engine/src/runtime/create-engine-runtime.ts new file mode 100644 index 00000000000..1f8bfa5d680 --- /dev/null +++ b/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; +} + +/** + * 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(); + const stepQueue = new InMemoryWorkQueue(); + 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()]); + }, + }; +} diff --git a/packages/@n8n/engine/src/runtime/index.ts b/packages/@n8n/engine/src/runtime/index.ts new file mode 100644 index 00000000000..f57838b94bd --- /dev/null +++ b/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'; diff --git a/packages/@n8n/engine/src/serve.ts b/packages/@n8n/engine/src/serve.ts index d0b3efd0b7b..7875038726a 100644 --- a/packages/@n8n/engine/src/serve.ts +++ b/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 { 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(); - const stepQueue = new InMemoryWorkQueue(); + 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 { await new Promise((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); }; diff --git a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts b/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts index ad94638b176..f31755f9a4e 100644 --- a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts +++ b/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 () => { diff --git a/packages/@n8n/engine/src/server/create-engine-server.ts b/packages/@n8n/engine/src/server/create-engine-server.ts index 0b82212f22b..3020cae8aa9 100644 --- a/packages/@n8n/engine/src/server/create-engine-server.ts +++ b/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; -} - -/** - * 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 }; } diff --git a/packages/@n8n/engine/src/server/index.ts b/packages/@n8n/engine/src/server/index.ts index 734d722b1a6..b68b301130a 100644 --- a/packages/@n8n/engine/src/server/index.ts +++ b/packages/@n8n/engine/src/server/index.ts @@ -1,2 +1 @@ export { createEngineServer } from './create-engine-server'; -export type { EngineServerDeps } from './create-engine-server'; diff --git a/packages/@n8n/engine/src/testing/__tests__/start-engine-server.integration.test.ts b/packages/@n8n/engine/src/testing/__tests__/start-engine-server.integration.test.ts index 2a00e01338e..9c6523ab104 100644 --- a/packages/@n8n/engine/src/testing/__tests__/start-engine-server.integration.test.ts +++ b/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; 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 () => { diff --git a/packages/@n8n/engine/src/testing/start-engine-server.ts b/packages/@n8n/engine/src/testing/start-engine-server.ts index 4cf1239a8a7..191d5528baa 100644 --- a/packages/@n8n/engine/src/testing/start-engine-server.ts +++ b/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; }> { - const { app } = createEngineServer(deps); + const { app } = createEngineServer(startExecution); const server = await new Promise((resolve, reject) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); diff --git a/packages/@n8n/node-engine-compatibility/package.json b/packages/@n8n/node-engine-compatibility/package.json index cd63bd261ab..fdd43e8ec9e 100644 --- a/packages/@n8n/node-engine-compatibility/package.json +++ b/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:" diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts index c1afe81cc9d..ee9b84a1e93 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts +++ b/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; 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(); - const stepQueue = new InMemoryWorkQueue(); let done!: () => void; const finished = new Promise((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 diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f46bf73b0fb..b2139ffd92c 100644 --- a/pnpm-lock.yaml +++ b/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