feat(core): Implement execution_start event handler (no-changelog) (#34795)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Iván Ovejero <ivov.src@gmail.com>

packages/@n8n/engine/AGENTS.md

@@ -6,8 +6,9 @@ We structure this package after the **Durable Scheduler modularity blueprint**
 ([Notion](https://app.notion.com/p/n8n/The-Durable-Scheduler-a-modularity-blueprint-39f5b6e0c94f8193a5fecca8306c4424),
 worked example: `packages/@n8n/scheduler`, enforced by its
 `src/__tests__/dependency-purity.test.ts`). The core idea: a **pure core that
-decides, with every effect handed in as a port**. Dependency arrows point one
-way — consumers depend on the engine; the engine core reaches for nothing.
+decides, with every effect handed in as an injected interface**. Dependency
+arrows point one way — consumers depend on the engine; the engine core reaches
+for nothing.
 
 ## Reality now: loose, but seam-aware
 
@@ -24,10 +25,10 @@ a deployable engine worker) without touching core logic.
   orchestration/policy. Must not open connections, bind an HTTP server, or read
   the environment. Everything external arrives via constructor args / a deps bag
   (the `createScheduler(deps)` pattern).
-- **Ports** — interfaces the core depends on, each defined in its own core
-  module beside a default/reference use: `AdmittanceService` (`admittance/`),
+- **Core interfaces** — interfaces the core depends on, each defined in its own
+  core module beside a default/reference use: `AdmittanceService` (`admittance/`),
   `WorkQueue` (`queue/`), `ExecutionStore` (`execution/`). Adapters implement
-  them; the core never imports the port from an adapter. Handed in at
+  them; the core never imports the interface from an adapter. Handed in at
   construction.
 - **Adapters (do)** — effectful implementations: `database/` (TypeORM entities,
   migrations, the Postgres `DataSource`, and `TypeOrmExecutionStore`), `queue/`
@@ -43,8 +44,8 @@ a deployable engine worker) without touching core logic.
 - Core modules (`graph`, `execution`, `admittance`) don't import `express`,
   `pg`, or `@n8n/typeorm`, don't construct a `DataSource`, and never import from
   `database/`. Persistence, queue, and HTTP are injected. The dependency arrow
-  is one-way: `database/` imports the port + domain types from the core, not the
-  reverse.
+  is one-way: `database/` imports the interface + domain types from the core, not
+  the reverse.
 - `@n8n/typeorm` / `pg` stay confined to `database/`.
 - `@n8n/config` + `@n8n/di` are used only at the serving/composition layer (for
   `EngineConfig`), never in core logic. (The blueprint flags `@n8n/config` as
@@ -59,6 +60,19 @@ a deployable engine worker) without touching core logic.
   `@n8n/scheduler` does) to enforce the allowlist. Until then, this doc is the
   intent.
 
+## Crash recovery: reconciliation, not transactions
+
+Handlers advance an execution through several separate writes — claim the
+execution, insert step rows, publish the next message — and we deliberately do
+**not** make that sequence atomic. Crashing partway through can leave partial
+state, such as an execution stuck `running` with no queued step.
+
+The intended answer is a reconciliation layer that detects crashed or stalled
+executions and drives recovery (CAT-2938), not transactions spanning stores and
+queues. So when you find a partial-write window: make the resulting state
+legible to reconciliation, and don't reach for a cross-store transaction. It's a
+recurring review question — this is the standing answer.
+
 ## Known deviations — the seams to clean up on decomposition
 
 - `DataSource` construction currently lives in `database/`; it is really a

packages/@n8n/engine/src/common/errors.ts

@@ -0,0 +1,12 @@
+/**
+ * Thrown when the engine reaches a code path that isn't implemented yet.
+ *
+ * Local because the engine avoids depending on `n8n-workflow`/`@n8n/core`.
+ * TODO(CAT-3798): move the engine's error types to a shared lib.
+ */
+export class UnimplementedError extends Error {
+	constructor(message: string) {
+		super(message);
+		this.name = 'UnimplementedError';
+	}
+}

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

@@ -1 +1,2 @@
 export type { JsonObject, JsonValue } from './json';
+export { UnimplementedError } from './errors';

packages/@n8n/engine/src/database/typeorm-execution-store.ts

@@ -1,7 +1,13 @@
 import type { Repository } from '@n8n/typeorm';
 
 import type { WorkflowExecution } from './entities';
-import type { ExecutionStore, NewExecutionRecord } from '../execution/execution-store';
+import {
+	ExecutionNotFoundError,
+	type ExecutionRecord,
+	type ExecutionStore,
+	type NewExecutionRecord,
+} from '../execution/execution-store';
+import type { ExecutionStatus } from '../execution/execution.types';
 
 /** TypeORM-backed `ExecutionStore` adapter. */
 export class TypeOrmExecutionStore implements ExecutionStore {
@@ -12,4 +18,15 @@ export class TypeOrmExecutionStore implements ExecutionStore {
 		await this.repo.save(execution);
 		return { id: execution.id };
 	}
+
+	async loadExecution(id: string): Promise<ExecutionRecord> {
+		const row = await this.repo.findOneBy({ id });
+		if (!row) throw new ExecutionNotFoundError(id);
+		return row;
+	}
+
+	async transitionStatus(id: string, from: ExecutionStatus, to: ExecutionStatus): Promise<boolean> {
+		const result = await this.repo.update({ id, status: from }, { status: to });
+		return result.affected === 1;
+	}
 }

packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts

@@ -0,0 +1,147 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import type { WorkflowGraph } from '../../graph';
+import type { StepMessage, WorkQueue } from '../../queue';
+import { ExecutionStartHandler } from '../execution-start-handler';
+import type { ExecutionRecord, ExecutionStore } from '../execution-store';
+import type { StepStore } from '../step-store';
+
+function makeExecutionStore(overrides: Partial<ExecutionStore> = {}): ExecutionStore {
+	return {
+		createExecution: vi.fn(),
+		loadExecution: vi.fn(),
+		transitionStatus: vi.fn().mockResolvedValue(true),
+		...overrides,
+	};
+}
+
+function makeStepQueue(): WorkQueue<StepMessage> {
+	return { publish: vi.fn(), start: vi.fn(), stop: vi.fn() };
+}
+
+function record(graph: WorkflowGraph): ExecutionRecord {
+	return {
+		id: 'exec-1',
+		workflowId: 'wf-1',
+		status: 'running',
+		mode: 'production',
+		graph,
+		triggerPayload: null,
+	};
+}
+
+describe('ExecutionStartHandler', () => {
+	it('records the trigger completed and enqueues a queued step per successor', async () => {
+		const graph: WorkflowGraph = {
+			nodes: [
+				{ id: 'trigger', name: 'T', type: 'trigger' },
+				{ id: 'a', name: 'A', type: 'v1-node' },
+				{ id: 'b', name: 'B', type: 'v1-node' },
+			],
+			edges: [
+				{ from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 0 },
+				{ from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 },
+			],
+		};
+		const executionStore = makeExecutionStore({
+			loadExecution: vi.fn().mockResolvedValue(record(graph)),
+		});
+		const createStep = vi
+			.fn()
+			.mockResolvedValueOnce({ id: 'step-trigger' })
+			.mockResolvedValueOnce({ id: 'step-a' })
+			.mockResolvedValueOnce({ id: 'step-b' });
+		const stepStore: StepStore = { createStep };
+		const stepQueue = makeStepQueue();
+		const handler = new ExecutionStartHandler(executionStore, stepStore, stepQueue);
+
+		await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' });
+
+		expect(executionStore.transitionStatus).toHaveBeenCalledWith('exec-1', 'queued', 'running');
+		// trigger recorded completed, then each successor recorded queued
+		expect(createStep).toHaveBeenNthCalledWith(1, {
+			executionId: 'exec-1',
+			nodeId: 'trigger',
+			status: 'completed',
+		});
+		expect(createStep).toHaveBeenNthCalledWith(2, {
+			executionId: 'exec-1',
+			nodeId: 'a',
+			status: 'queued',
+		});
+		expect(createStep).toHaveBeenNthCalledWith(3, {
+			executionId: 'exec-1',
+			nodeId: 'b',
+			status: 'queued',
+		});
+		// step:ready references the queued step-record ids
+		expect(stepQueue.publish).toHaveBeenNthCalledWith(1, {
+			type: 'step:ready',
+			executionId: 'exec-1',
+			stepId: 'step-a',
+		});
+		expect(stepQueue.publish).toHaveBeenNthCalledWith(2, {
+			type: 'step:ready',
+			executionId: 'exec-1',
+			stepId: 'step-b',
+		});
+	});
+
+	it('is a no-op when the execution cannot be claimed (duplicate delivery)', async () => {
+		const executionStore = makeExecutionStore({
+			transitionStatus: vi.fn().mockResolvedValue(false),
+		});
+		const stepStore: StepStore = { createStep: vi.fn() };
+		const stepQueue = makeStepQueue();
+		const handler = new ExecutionStartHandler(executionStore, stepStore, stepQueue);
+
+		await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' });
+
+		expect(executionStore.loadExecution).not.toHaveBeenCalled();
+		expect(stepStore.createStep).not.toHaveBeenCalled();
+		expect(stepQueue.publish).not.toHaveBeenCalled();
+	});
+
+	it('fails the execution when the graph has no trigger node', async () => {
+		const graph: WorkflowGraph = { nodes: [{ id: 'a', name: 'A', type: 'v1-node' }], edges: [] };
+		const executionStore = makeExecutionStore({
+			loadExecution: vi.fn().mockResolvedValue(record(graph)),
+		});
+		const stepStore: StepStore = { createStep: vi.fn() };
+		const stepQueue = makeStepQueue();
+		const handler = new ExecutionStartHandler(executionStore, stepStore, stepQueue);
+
+		await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' });
+
+		expect(executionStore.transitionStatus).toHaveBeenCalledWith('exec-1', 'running', 'failed');
+		expect(stepStore.createStep).not.toHaveBeenCalled();
+		expect(stepQueue.publish).not.toHaveBeenCalled();
+	});
+
+	it('records the trigger but enqueues nothing for a trigger with no successors', async () => {
+		const graph: WorkflowGraph = {
+			nodes: [{ id: 'trigger', name: 'T', type: 'trigger' }],
+			edges: [],
+		};
+		const executionStore = makeExecutionStore({
+			loadExecution: vi.fn().mockResolvedValue(record(graph)),
+		});
+		const createStep = vi.fn().mockResolvedValue({ id: 'step-trigger' });
+		const stepStore: StepStore = { createStep };
+		const stepQueue = makeStepQueue();
+		const handler = new ExecutionStartHandler(executionStore, stepStore, stepQueue);
+
+		await handler.handle({ type: 'execution:enqueued', executionId: 'exec-1' });
+
+		// claimed (queued -> running) but not failed
+		expect(executionStore.transitionStatus).toHaveBeenCalledWith('exec-1', 'queued', 'running');
+		expect(executionStore.transitionStatus).not.toHaveBeenCalledWith('exec-1', 'running', 'failed');
+		expect(createStep).toHaveBeenCalledTimes(1);
+		expect(createStep).toHaveBeenCalledWith({
+			executionId: 'exec-1',
+			nodeId: 'trigger',
+			status: 'completed',
+		});
+		expect(stepQueue.publish).not.toHaveBeenCalled();
+	});
+});

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

@@ -0,0 +1,138 @@
+import type { DataSource } from '@n8n/typeorm';
+import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql';
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
+
+import { AllowAllAdmittance } from '../../admittance';
+import {
+	createDataSource,
+	TypeOrmExecutionStore,
+	TypeOrmStepStore,
+	WorkflowExecution,
+	WorkflowStepExecution,
+} from '../../database';
+import type { WorkflowGraph } from '../../graph';
+import {
+	InMemoryWorkQueue,
+	type OrchestrationMessage,
+	type StepMessage,
+	type WorkQueue,
+} from '../../queue';
+import { ExecutionStartHandler } from '../execution-start-handler';
+import { OrchestrationWorker } from '../orchestration-worker';
+import { StartExecutionService } from '../start-execution.service';
+
+const graph: WorkflowGraph = {
+	nodes: [
+		{ id: 'trigger', name: 'Manual Trigger', type: 'trigger' },
+		{ id: 'step-a', name: 'A', type: 'v1-node' },
+	],
+	edges: [{ from: 'trigger', to: 'step-a', outputIndex: 0, inputIndex: 0 }],
+};
+
+describe('execution start (integration)', () => {
+	let container: StartedPostgreSqlContainer;
+	let dataSource: DataSource;
+
+	beforeAll(async () => {
+		container = await new PostgreSqlContainer('postgres:18-alpine').start();
+		dataSource = createDataSource(container.getConnectionUri());
+		await dataSource.initialize();
+		await dataSource.runMigrations();
+	}, 120_000);
+
+	afterAll(async () => {
+		if (dataSource?.isInitialized) await dataSource.destroy();
+		if (container) await container.stop();
+	});
+
+	function stores() {
+		return {
+			executionStore: new TypeOrmExecutionStore(dataSource.getRepository(WorkflowExecution)),
+			stepStore: new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution)),
+		};
+	}
+
+	it('runs the execution, records the trigger completed + first step queued, enqueues step:ready', async () => {
+		const { executionStore, stepStore } = stores();
+		const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
+		const stepQueue = new InMemoryWorkQueue<StepMessage>();
+		const worker = new OrchestrationWorker(
+			orchestrationQueue,
+			new ExecutionStartHandler(executionStore, stepStore, stepQueue),
+		);
+		worker.start();
+		const startExecution = new StartExecutionService(
+			new AllowAllAdmittance(),
+			executionStore,
+			orchestrationQueue,
+		);
+
+		// Stand in for the (not yet built) step worker: consuming the step queue is
+		// both the assertion target and the signal that orchestration finished.
+		const readySteps: StepMessage[] = [];
+		let firstReady!: () => void;
+		const ready = new Promise<void>((resolve) => (firstReady = resolve));
+		stepQueue.start(async (message) => {
+			readySteps.push(message);
+			firstReady();
+			await Promise.resolve();
+		});
+
+		const { executionId } = await startExecution.start({
+			workflowId: 'wf-1',
+			graph,
+			triggerPayload: null,
+		});
+		await ready;
+
+		const row = await dataSource
+			.getRepository(WorkflowExecution)
+			.findOneByOrFail({ id: executionId });
+		expect(row.status).toBe('running');
+
+		const steps = await dataSource
+			.getRepository(WorkflowStepExecution)
+			.find({ where: { executionId } });
+		const triggerStep = steps.find((s) => s.nodeId === 'trigger');
+		const firstStep = steps.find((s) => s.nodeId === 'step-a');
+		expect(triggerStep?.status).toBe('completed');
+		expect(firstStep?.status).toBe('queued');
+
+		// step:ready references the durable step-record id, not the node id.
+		expect(readySteps).toEqual([{ type: 'step:ready', executionId, stepId: firstStep?.id }]);
+
+		await worker.stop();
+		await stepQueue.stop();
+	});
+
+	it('is idempotent across duplicate execution:enqueued deliveries', async () => {
+		const { executionStore, stepStore } = stores();
+		const publish = vi.fn();
+		const stepQueue: WorkQueue<StepMessage> = { publish, start: vi.fn(), stop: vi.fn() };
+		const handler = new ExecutionStartHandler(executionStore, stepStore, stepQueue);
+
+		const { id: executionId } = await executionStore.createExecution({
+			workflowId: 'wf-2',
+			status: 'queued',
+			mode: 'production',
+			graph,
+			triggerPayload: null,
+		});
+
+		// Delivered twice, both awaited — the CAS is what makes the second a no-op.
+		const event = { type: 'execution:enqueued', executionId } as const;
+		await handler.handle(event);
+		await handler.handle(event);
+
+		const row = await dataSource
+			.getRepository(WorkflowExecution)
+			.findOneByOrFail({ id: executionId });
+		expect(row.status).toBe('running');
+
+		const steps = await dataSource
+			.getRepository(WorkflowStepExecution)
+			.find({ where: { executionId } });
+		expect(steps).toHaveLength(2); // trigger (completed) + step-a (queued)
+		expect(publish).toHaveBeenCalledTimes(1);
+	});
+});

packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts

@@ -2,7 +2,7 @@ import { describe, expect, it, vi } from 'vitest';
 
 import { AdmittanceRejectedError, type AdmittanceService } from '../../admittance';
 import type { WorkflowGraph } from '../../graph';
-import type { WorkQueue, WorkQueueMessage } from '../../queue';
+import type { OrchestrationMessage, WorkQueue } from '../../queue';
 import type { ExecutionStore } from '../execution-store';
 import { StartExecutionService } from '../start-execution.service';
 
@@ -11,24 +11,27 @@ const sampleGraph: WorkflowGraph = {
 	edges: [],
 };
 
+function makeQueue(): WorkQueue<OrchestrationMessage> {
+	return { publish: vi.fn(), start: vi.fn(), stop: vi.fn() };
+}
+
+function makeStore(overrides: Partial<ExecutionStore> = {}): ExecutionStore {
+	return {
+		createExecution: vi.fn().mockResolvedValue({ id: 'exec-id-1' }),
+		loadExecution: vi.fn(),
+		transitionStatus: vi.fn().mockResolvedValue(true),
+		...overrides,
+	};
+}
+
 describe('StartExecutionService', () => {
 	it('admits, persists a queued execution, publishes execution:enqueued, returns id', async () => {
 		const admittance: AdmittanceService = {
 			evaluate: vi.fn().mockResolvedValue({ accept: true }),
 		};
-		const executionStore: ExecutionStore = {
-			createExecution: vi.fn().mockResolvedValue({ id: 'exec-id-1' }),
-		};
-		const messages: WorkQueueMessage[] = [];
-		const queue: WorkQueue = {
-			publish: vi.fn().mockImplementation(
-				// eslint-disable-next-line @typescript-eslint/require-await -- mock impl
-				async (m: WorkQueueMessage) => {
-					messages.push(m);
-				},
-			),
-		};
-		const service = new StartExecutionService(admittance, executionStore, queue);
+		const store = makeStore();
+		const queue = makeQueue();
+		const service = new StartExecutionService(admittance, store, queue);
 
 		const result = await service.start({
 			workflowId: 'wf-1',
@@ -38,29 +41,30 @@ describe('StartExecutionService', () => {
 
 		expect(result.executionId).toBe('exec-id-1');
 		expect(admittance.evaluate).toHaveBeenCalledWith({ workflowId: 'wf-1' });
-		expect(executionStore.createExecution).toHaveBeenCalledWith({
+		expect(store.createExecution).toHaveBeenCalledWith({
 			workflowId: 'wf-1',
 			status: 'queued',
 			mode: 'production',
 			graph: sampleGraph,
 			triggerPayload: { hello: 'world' },
 		});
-		expect(messages).toEqual([{ type: 'execution:enqueued', executionId: 'exec-id-1' }]);
+		expect(queue.publish).toHaveBeenCalledWith({
+			type: 'execution:enqueued',
+			executionId: 'exec-id-1',
+		});
 	});
 
 	it('defaults mode to production and triggerPayload to null', async () => {
 		const admittance: AdmittanceService = {
 			evaluate: vi.fn().mockResolvedValue({ accept: true }),
 		};
-		const executionStore: ExecutionStore = {
-			createExecution: vi.fn().mockResolvedValue({ id: 'exec-id-1' }),
-		};
-		const queue: WorkQueue = { publish: vi.fn().mockResolvedValue(undefined) };
-		const service = new StartExecutionService(admittance, executionStore, queue);
+		const store = makeStore();
+		const queue = makeQueue();
+		const service = new StartExecutionService(admittance, store, queue);
 
 		await service.start({ workflowId: 'wf-1', graph: sampleGraph });
 
-		expect(executionStore.createExecution).toHaveBeenCalledWith(
+		expect(store.createExecution).toHaveBeenCalledWith(
 			expect.objectContaining({ mode: 'production', triggerPayload: null }),
 		);
 	});
@@ -69,15 +73,15 @@ describe('StartExecutionService', () => {
 		const admittance: AdmittanceService = {
 			evaluate: vi.fn().mockResolvedValue({ accept: false, reason: 'queue-full' }),
 		};
-		const executionStore: ExecutionStore = { createExecution: vi.fn() };
-		const queue: WorkQueue = { publish: vi.fn() };
-		const service = new StartExecutionService(admittance, executionStore, queue);
+		const store = makeStore();
+		const queue = makeQueue();
+		const service = new StartExecutionService(admittance, store, queue);
 
 		await expect(service.start({ workflowId: 'wf-1', graph: sampleGraph })).rejects.toBeInstanceOf(
 			AdmittanceRejectedError,
 		);
 
-		expect(executionStore.createExecution).not.toHaveBeenCalled();
+		expect(store.createExecution).not.toHaveBeenCalled();
 		expect(queue.publish).not.toHaveBeenCalled();
 	});
 });

packages/@n8n/engine/src/execution/execution-start-handler.ts

@@ -0,0 +1,58 @@
+import { findTriggerNode, getSuccessorNodeIds } from '../graph';
+import type { ExecutionEnqueuedEvent, StepMessage, WorkQueue } from '../queue';
+import type { ExecutionStore } from './execution-store';
+import type { StepStore } from './step-store';
+
+/**
+ * Handles the `execution:enqueued` orchestration event: claims the execution
+ * (`queued → running`), records the trigger as a completed step, and plans the
+ * first step(s).
+ */
+export class ExecutionStartHandler {
+	constructor(
+		private readonly executionStore: ExecutionStore,
+		private readonly stepStore: StepStore,
+		private readonly stepQueue: WorkQueue<StepMessage>,
+	) {}
+
+	async handle(event: ExecutionEnqueuedEvent): Promise<void> {
+		// Claim via CAS so a duplicate/redelivered event is a no-op.
+		const claimed = await this.executionStore.transitionStatus(
+			event.executionId,
+			'queued',
+			'running',
+		);
+		if (!claimed) return;
+
+		const execution = await this.executionStore.loadExecution(event.executionId);
+
+		const trigger = findTriggerNode(execution.graph);
+		if (!trigger) {
+			// Malformed graph — no entry point to run.
+			await this.executionStore.transitionStatus(event.executionId, 'running', 'failed');
+			return;
+		}
+
+		// The trigger's output was captured at execution start; record it as
+		// already completed so successors can treat it as a satisfied predecessor.
+		await this.stepStore.createStep({
+			executionId: event.executionId,
+			nodeId: trigger.id,
+			status: 'completed',
+		});
+
+		const successorNodeIds = getSuccessorNodeIds(execution.graph, trigger.id);
+		for (const nodeId of successorNodeIds) {
+			const { id: stepId } = await this.stepStore.createStep({
+				executionId: event.executionId,
+				nodeId,
+				status: 'queued',
+			});
+			await this.stepQueue.publish({
+				type: 'step:ready',
+				executionId: event.executionId,
+				stepId,
+			});
+		}
+	}
+}

packages/@n8n/engine/src/execution/execution-store.ts

@@ -11,10 +11,35 @@ export interface NewExecutionRecord {
 	triggerPayload: JsonObject | null;
 }
 
-/**
- * The persistence store for executions.
- */
+/** A full execution record. */
+export interface ExecutionRecord {
+	id: string;
+	workflowId: string;
+	status: ExecutionStatus;
+	mode: ExecutionMode;
+	graph: WorkflowGraph;
+	triggerPayload: JsonObject | null;
+}
+
+/** Thrown by `loadExecution` when no execution exists for the given id. */
+export class ExecutionNotFoundError extends Error {
+	constructor(readonly executionId: string) {
+		super(`Execution not found: ${executionId}`);
+		this.name = 'ExecutionNotFoundError';
+	}
+}
+
+/** Persistence interface for executions. */
 export interface ExecutionStore {
 	/** Persist a new execution record; returns its generated id. */
 	createExecution(record: NewExecutionRecord): Promise<{ id: string }>;
+
+	/** Load a full execution by id. Throws `ExecutionNotFoundError` if absent. */
+	loadExecution(id: string): Promise<ExecutionRecord>;
+
+	/**
+	 * Compare-and-set status transition. Returns `true` iff this call performed
+	 * the transition, so duplicate/redelivered events are handled idempotently.
+	 */
+	transitionStatus(id: string, from: ExecutionStatus, to: ExecutionStatus): Promise<boolean>;
 }

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

@@ -4,5 +4,8 @@ export type {
 	StartExecutionResult,
 } from './start-execution.service';
 export type { ExecutionMode, ExecutionStatus, StepStatus } from './execution.types';
-export type { ExecutionStore, NewExecutionRecord } from './execution-store';
+export { ExecutionNotFoundError } from './execution-store';
+export type { ExecutionRecord, ExecutionStore, NewExecutionRecord } from './execution-store';
 export type { NewStepRecord, StepStore } from './step-store';
+export { ExecutionStartHandler } from './execution-start-handler';
+export { OrchestrationWorker } from './orchestration-worker';

packages/@n8n/engine/src/execution/orchestration-worker.ts

@@ -0,0 +1,31 @@
+import { UnimplementedError } from '../common';
+import type { OrchestrationMessage, WorkQueue } from '../queue';
+import type { ExecutionStartHandler } from './execution-start-handler';
+
+/**
+ * Consumes the orchestration queue and routes each message to its handler.
+ */
+export class OrchestrationWorker {
+	constructor(
+		private readonly orchestrationQueue: WorkQueue<OrchestrationMessage>,
+		private readonly startHandler: ExecutionStartHandler,
+	) {}
+
+	start(): void {
+		this.orchestrationQueue.start(async (message) => {
+			switch (message.type) {
+				case 'execution:enqueued':
+					await this.startHandler.handle(message);
+					break;
+				default:
+					throw new UnimplementedError(
+						`orchestration worker received an unimplemented message type: ${String(message.type)}`,
+					);
+			}
+		});
+	}
+
+	async stop(): Promise<void> {
+		await this.orchestrationQueue.stop();
+	}
+}

packages/@n8n/engine/src/execution/start-execution.service.ts

@@ -1,7 +1,7 @@
 import { AdmittanceRejectedError, type AdmittanceService } from '../admittance';
 import type { JsonObject } from '../common';
 import type { WorkflowGraph } from '../graph';
-import type { WorkQueue } from '../queue';
+import type { OrchestrationMessage, WorkQueue } from '../queue';
 import type { ExecutionStore } from './execution-store';
 import type { ExecutionMode } from './execution.types';
 
@@ -20,7 +20,7 @@ export class StartExecutionService {
 	constructor(
 		private readonly admittance: AdmittanceService,
 		private readonly executionStore: ExecutionStore,
-		private readonly workQueue: WorkQueue,
+		private readonly workQueue: WorkQueue<OrchestrationMessage>,
 	) {}
 
 	async start(request: StartExecutionRequest): Promise<StartExecutionResult> {

packages/@n8n/engine/src/execution/step-store.ts

@@ -7,9 +7,7 @@ export interface NewStepRecord {
 	status: StepStatus;
 }
 
-/**
- * Persistence port for step records.
- */
+/** Persistence interface for step records. */
 export interface StepStore {
 	/** Persist a new step record; returns its generated id. */
 	createStep(record: NewStepRecord): Promise<{ id: string }>;

packages/@n8n/engine/src/graph/__tests__/workflow-graph-queries.test.ts

@@ -0,0 +1,45 @@
+import { describe, expect, it } from 'vitest';
+
+import type { WorkflowGraph } from '../workflow-graph';
+import { findTriggerNode, getSuccessorNodeIds } from '../workflow-graph-queries';
+
+const graph: WorkflowGraph = {
+	nodes: [
+		{ id: 'trigger', name: 'Trigger', type: 'trigger' },
+		{ id: 'a', name: 'A', type: 'v1-node' },
+		{ id: 'b', name: 'B', type: 'v1-node' },
+	],
+	edges: [
+		{ from: 'trigger', to: 'a', outputIndex: 0, inputIndex: 0 },
+		{ from: 'trigger', to: 'b', outputIndex: 0, inputIndex: 0 },
+		// Same target from a second output slot — must not be reported twice.
+		{ from: 'trigger', to: 'a', outputIndex: 1, inputIndex: 0 },
+		{ from: 'a', to: 'b', outputIndex: 0, inputIndex: 0 },
+	],
+};
+
+describe('findTriggerNode', () => {
+	it('returns the trigger node', () => {
+		expect(findTriggerNode(graph)?.id).toBe('trigger');
+	});
+
+	it('returns undefined when there is no trigger', () => {
+		expect(
+			findTriggerNode({ nodes: [{ id: 'a', name: 'A', type: 'v1-node' }], edges: [] }),
+		).toBeUndefined();
+	});
+});
+
+describe('getSuccessorNodeIds', () => {
+	it('returns forward successors in edge order, de-duplicated', () => {
+		expect(getSuccessorNodeIds(graph, 'trigger')).toEqual(['a', 'b']);
+	});
+
+	it('returns an empty array for a node with no outgoing edges', () => {
+		expect(getSuccessorNodeIds(graph, 'b')).toEqual([]);
+	});
+
+	it('returns an empty array for an unknown node', () => {
+		expect(getSuccessorNodeIds(graph, 'nope')).toEqual([]);
+	});
+});

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

@@ -5,3 +5,4 @@ export type {
 	StepType,
 	WorkflowGraph,
 } from './workflow-graph';
+export { findTriggerNode, getSuccessorNodeIds } from './workflow-graph-queries';

packages/@n8n/engine/src/graph/workflow-graph-queries.ts

@@ -0,0 +1,21 @@
+import type { GraphNode, WorkflowGraph } from './workflow-graph';
+
+/** The trigger node (the execution's entry point), or `undefined` if there is none. */
+export function findTriggerNode(graph: WorkflowGraph): GraphNode | undefined {
+	return graph.nodes.find((node) => node.type === 'trigger');
+}
+
+/**
+ * Ids of the nodes directly downstream of `nodeId`, de-duplicated, in edge order.
+ *
+ * TODO(CAT-3854): validate edge endpoints against `nodes`.
+ */
+export function getSuccessorNodeIds(graph: WorkflowGraph, nodeId: string): string[] {
+	const successors: string[] = [];
+	for (const edge of graph.edges) {
+		if (edge.from === nodeId && !successors.includes(edge.to)) {
+			successors.push(edge.to);
+		}
+	}
+	return successors;
+}

packages/@n8n/engine/src/graph/workflow-graph.ts

@@ -29,11 +29,15 @@ export interface GraphEdge {
 	from: string;
 	/** Target node id. */
 	to: string;
-	/** Which output slot of `from` feeds this edge — multi-output nodes only. */
-	outputIndex?: number;
-	/** Which input slot of `to` this edge feeds — multi-input nodes (e.g. Merge). */
-	inputIndex?: number;
-	/** Closes a cycle (loop iteration). */
+	/** Which output slot of `from` feeds this edge; `0` unless `from` has several outputs. */
+	outputIndex: number;
+	/** Which input slot of `to` this edge feeds; `0` unless `to` has several inputs (e.g. Merge). */
+	inputIndex: number;
+	/**
+	 * Closes a cycle (loop iteration).
+	 *
+	 * TODO(CAT-3854): validate that back-edges really close loops.
+	 */
 	isBackEdge?: boolean;
 }
 

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

@@ -29,12 +29,16 @@ export type {
 export { InMemoryWorkQueue } from './queue';
 export type {
 	ExecutionEnqueuedEvent,
+	OrchestrationMessage,
+	StepMessage,
+	StepReadyEvent,
 	WorkQueue,
-	WorkQueueMessage,
 } from './queue';
 
+export { ExecutionNotFoundError, ExecutionStartHandler, OrchestrationWorker } from './execution';
 export type {
 	ExecutionMode,
+	ExecutionRecord,
 	ExecutionStatus,
 	ExecutionStore,
 	NewExecutionRecord,

packages/@n8n/engine/src/queue/__tests__/in-memory-work-queue.test.ts

@@ -0,0 +1,90 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { InMemoryWorkQueue } from '../in-memory-work-queue';
+
+interface Msg {
+	type: 'x';
+	n: number;
+}
+
+/**
+ * A consumer that records what it receives and resolves `complete` once
+ * `expected` messages have arrived — dispatch is asynchronous, so tests need
+ * their own signal rather than a hook into the queue.
+ */
+function collect(expected: number) {
+	const received: Msg[] = [];
+	let done!: () => void;
+	const complete = new Promise<void>((resolve) => (done = resolve));
+	const handler = async (message: Msg) => {
+		received.push(message);
+		if (received.length >= expected) done();
+		await Promise.resolve();
+	};
+	return { received, complete, handler };
+}
+
+describe('InMemoryWorkQueue', () => {
+	it('dispatches published messages to a consumer, in order', async () => {
+		const queue = new InMemoryWorkQueue<Msg>();
+		const { received, complete, handler } = collect(2);
+		queue.start(handler);
+
+		await queue.publish({ type: 'x', n: 1 });
+		await queue.publish({ type: 'x', n: 2 });
+		await complete;
+
+		expect(received).toEqual([
+			{ type: 'x', n: 1 },
+			{ type: 'x', n: 2 },
+		]);
+	});
+
+	it('dispatches messages published before a consumer registers', async () => {
+		const queue = new InMemoryWorkQueue<Msg>();
+		await queue.publish({ type: 'x', n: 1 });
+
+		const { received, complete, handler } = collect(1);
+		queue.start(handler);
+		await complete;
+
+		expect(received).toEqual([{ type: 'x', n: 1 }]);
+	});
+
+	it('keeps dispatching after a consumer throws', async () => {
+		vi.spyOn(console, 'error').mockImplementation(() => {});
+		const queue = new InMemoryWorkQueue<Msg>();
+		const { received, complete, handler } = collect(2);
+		queue.start(async (message) => {
+			await handler(message);
+			if (message.n === 1) throw new Error('boom');
+		});
+
+		await queue.publish({ type: 'x', n: 1 });
+		await queue.publish({ type: 'x', n: 2 });
+		await complete;
+
+		expect(received).toEqual([
+			{ type: 'x', n: 1 },
+			{ type: 'x', n: 2 },
+		]);
+	});
+
+	it('stop awaits the in-flight message', async () => {
+		const queue = new InMemoryWorkQueue<Msg>();
+		const { received, handler } = collect(1);
+		queue.start(handler);
+
+		await queue.publish({ type: 'x', n: 1 });
+		await queue.stop();
+
+		expect(received).toEqual([{ type: 'x', n: 1 }]);
+	});
+
+	it('stop resolves when messages were published without a consumer', async () => {
+		const queue = new InMemoryWorkQueue<Msg>();
+		await queue.publish({ type: 'x', n: 1 });
+
+		await expect(queue.stop()).resolves.toBeUndefined();
+	});
+});

packages/@n8n/engine/src/queue/in-memory-work-queue.ts

@@ -1,17 +1,65 @@
-import type { WorkQueue, WorkQueueMessage } from './work-queue.types';
+import type { WorkQueue } from './work-queue.types';
 
 /**
- * Default `WorkQueue` impl: appends published messages to an in-memory array.
- * Useful for tests and as the default until a Redis-backed impl lands.
- *
- * `messages` is exposed deliberately so tests can assert on what was
- * published without any extra wiring.
+ * In-memory `WorkQueue`. When a handler is registered it dispatches messages
+ * sequentially on a microtask.
  */
-export class InMemoryWorkQueue implements WorkQueue {
-	readonly messages: WorkQueueMessage[] = [];
+export class InMemoryWorkQueue<TMessage> implements WorkQueue<TMessage> {
+	private pending: TMessage[] = [];
 
-	// eslint-disable-next-line @typescript-eslint/require-await -- satisfies async interface; impl has no async work
-	async publish(message: WorkQueueMessage): Promise<void> {
-		this.messages.push(message);
+	private handler: ((message: TMessage) => Promise<void>) | undefined;
+
+	private dispatching = false;
+
+	private idleWaiters: Array<() => void> = [];
+
+	// eslint-disable-next-line @typescript-eslint/require-await -- satisfies async interface; dispatch is scheduled, not awaited
+	async publish(message: TMessage): Promise<void> {
+		this.pending.push(message);
+		this.startDispatchLoop();
+	}
+
+	start(handler: (message: TMessage) => Promise<void>): void {
+		this.handler = handler;
+		this.startDispatchLoop();
+	}
+
+	async stop(): Promise<void> {
+		this.handler = undefined;
+		// Only in-flight work is awaited; anything still queued has no consumer to
+		// dispatch it, so waiting on it would never return.
+		if (!this.dispatching) return;
+		await new Promise<void>((resolve) => this.idleWaiters.push(resolve));
+	}
+
+	/**
+	 * Starts dispatching unless it is already running, there is no consumer, or
+	 * there is nothing queued. Safe to call on every publish.
+	 */
+	private startDispatchLoop(): void {
+		if (this.dispatching || !this.handler || this.pending.length === 0) return;
+		this.dispatching = true;
+		this.dispatchPending().catch((error: unknown) => {
+			console.error('engine: work queue dispatch failed', error);
+		});
+	}
+
+	private async dispatchPending(): Promise<void> {
+		try {
+			while (this.handler && this.pending.length > 0) {
+				const [message] = this.pending.splice(0, 1);
+				try {
+					await this.handler(message);
+				} catch (error) {
+					// Contained per message: a failed one must not strand those behind it.
+					console.error('engine: work queue handler failed', error);
+				}
+			}
+		} finally {
+			this.dispatching = false;
+			const waiters = this.idleWaiters;
+			this.idleWaiters = [];
+			for (const resolve of waiters) resolve();
+		}
 	}
 }

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

@@ -1,6 +1,8 @@
 export type {
 	ExecutionEnqueuedEvent,
+	OrchestrationMessage,
+	StepMessage,
+	StepReadyEvent,
 	WorkQueue,
-	WorkQueueMessage,
 } from './work-queue.types';
 export { InMemoryWorkQueue } from './in-memory-work-queue';

packages/@n8n/engine/src/queue/work-queue.types.ts

@@ -1,7 +1,10 @@
 /**
- * The work queue for engine tasks.
+ * Engine work queues.
  *
- * TODO: add additional message types and functionality as we build the engine components.
+ * Two logical queues are used so a flood of step events can't starve
+ * orchestration (or vice versa): an **orchestration queue** (execution/step
+ * lifecycle planning) and a **step queue** (steps ready to execute). `WorkQueue`
+ * is generic over its message type; adapters implement it per queue.
  */
 
 export interface ExecutionEnqueuedEvent {
@@ -9,8 +12,25 @@ export interface ExecutionEnqueuedEvent {
 	executionId: string;
 }
 
-export type WorkQueueMessage = ExecutionEnqueuedEvent;
+/** Messages consumed by the orchestration worker. */
+export type OrchestrationMessage = ExecutionEnqueuedEvent;
 
-export interface WorkQueue {
-	publish(message: WorkQueueMessage): Promise<void>;
+export interface StepReadyEvent {
+	type: 'step:ready';
+	executionId: string;
+	stepId: string;
+}
+
+/** Messages consumed by the step worker (CAT-2870). */
+export type StepMessage = StepReadyEvent;
+
+export interface WorkQueue<TMessage> {
+	publish(message: TMessage): Promise<void>;
+	/** Register the single consumer; dispatch begins for queued and future messages. */
+	start(handler: (message: TMessage) => Promise<void>): void;
+	/**
+	 * Stop consuming; awaits any in-flight handler and leaves queued messages
+	 * unconsumed. Never blocks on work no longer being dispatched.
+	 */
+	stop(): Promise<void>;
 }

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

@@ -3,8 +3,16 @@ import { Container } from '@n8n/di';
 import type { DataSource } from '@n8n/typeorm';
 
 import { AllowAllAdmittance } from './admittance';
-import { createDataSource } from './database';
+import {
+	createDataSource,
+	TypeOrmExecutionStore,
+	TypeOrmStepStore,
+	WorkflowExecution,
+	WorkflowStepExecution,
+} from './database';
+import { ExecutionStartHandler, OrchestrationWorker } from './execution';
 import { InMemoryWorkQueue } from './queue';
+import type { OrchestrationMessage, StepMessage } from './queue';
 import { createEngineServer } from './server';
 
 async function main(): Promise<void> {
@@ -21,13 +29,25 @@ async function main(): Promise<void> {
 		);
 	}
 
+	const orchestrationQueue = new InMemoryWorkQueue<OrchestrationMessage>();
+	// Published to but not yet consumed — the step worker is CAT-2870. Queued
+	// messages dispatch as soon as it registers.
+	const stepQueue = new InMemoryWorkQueue<StepMessage>();
+
+	let worker: OrchestrationWorker | undefined;
+	if (dataSource) {
+		const executionStore = new TypeOrmExecutionStore(dataSource.getRepository(WorkflowExecution));
+		const stepStore = new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution));
+		worker = new OrchestrationWorker(
+			orchestrationQueue,
+			new ExecutionStartHandler(executionStore, stepStore, stepQueue),
+		);
+		worker.start();
+	}
+
 	const { app } = createEngineServer(
 		dataSource
-			? {
-					dataSource,
-					admittance: new AllowAllAdmittance(),
-					workQueue: new InMemoryWorkQueue(),
-				}
+			? { dataSource, admittance: new AllowAllAdmittance(), workQueue: orchestrationQueue }
 			: undefined,
 	);
 
@@ -43,6 +63,7 @@ async function main(): Promise<void> {
 		await new Promise<void>((resolve, reject) => {
 			server.close((error) => (error ? reject(error) : resolve()));
 		});
+		if (worker) await worker.stop();
 		if (dataSource?.isInitialized) await dataSource.destroy();
 		process.exit(0);
 	};

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

@@ -1,12 +1,12 @@
 import type { DataSource } from '@n8n/typeorm';
 import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql';
 import request from 'supertest';
-import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { AllowAllAdmittance } from '../../admittance';
 import { createDataSource, WorkflowExecution } from '../../database';
 import type { WorkflowGraph } from '../../graph';
-import { InMemoryWorkQueue } from '../../queue';
+import type { OrchestrationMessage, WorkQueue } from '../../queue';
 import { startEngineServer } from '../../testing/start-engine-server';
 
 const sampleGraph: WorkflowGraph = {
@@ -17,7 +17,7 @@ const sampleGraph: WorkflowGraph = {
 describe('POST /api/workflow-executions (integration)', () => {
 	let container: StartedPostgreSqlContainer;
 	let dataSource: DataSource;
-	let workQueue: InMemoryWorkQueue;
+	let workQueue: WorkQueue<OrchestrationMessage>;
 	let url: string;
 	let stop: () => Promise<void>;
 
@@ -29,7 +29,7 @@ describe('POST /api/workflow-executions (integration)', () => {
 	}, 120_000);
 
 	beforeEach(async () => {
-		workQueue = new InMemoryWorkQueue();
+		workQueue = { publish: vi.fn(), start: vi.fn(), stop: vi.fn() };
 		({ url, stop } = await startEngineServer({
 			dataSource,
 			admittance: new AllowAllAdmittance(),
@@ -67,7 +67,7 @@ describe('POST /api/workflow-executions (integration)', () => {
 		expect(row.graph).toEqual(sampleGraph);
 		expect(row.triggerPayload).toEqual({ hello: 'world' });
 
-		expect(workQueue.messages).toEqual([{ type: 'execution:enqueued', executionId }]);
+		expect(workQueue.publish).toHaveBeenCalledWith({ type: 'execution:enqueued', executionId });
 	});
 
 	it('rejects an invalid body with 400', async () => {

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

@@ -4,13 +4,13 @@ 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 { WorkQueue } from '../queue';
+import type { OrchestrationMessage, WorkQueue } from '../queue';
 import { createWorkflowExecutionsRouter } from './routes/workflow-executions';
 
 export interface EngineServerDeps {
 	dataSource: DataSource;
 	admittance: AdmittanceService;
-	workQueue: WorkQueue;
+	workQueue: WorkQueue<OrchestrationMessage>;
 }
 
 /**

packages/@n8n/engine/src/server/routes/workflow-executions.ts

@@ -30,8 +30,10 @@ const GraphNodeSchema = z.object({
 const GraphEdgeSchema = z.object({
 	from: z.string(),
 	to: z.string(),
-	outputIndex: z.number().int().nonnegative().optional(),
-	inputIndex: z.number().int().nonnegative().optional(),
+	// Defaulted rather than optional so callers can omit the common 0/0 case while
+	// the engine always sees a fully-populated edge.
+	outputIndex: z.number().int().nonnegative().default(0),
+	inputIndex: z.number().int().nonnegative().default(0),
 	isBackEdge: z.boolean().optional(),
 });