perf(engine): Cut database round trips on the step settle and run paths

Each queue on the data plane handles one message at a time, so every
sequential round trip adds to the time per execution.

- createSteps: move the failed-step test into the INSERT. The test still
  runs as a statement after the execution-row lock, so it sees a failure
  that commits while the insert waits for the lock. 5 round trips become 4.
- step:settled: read hasFailedSteps in parallel with loadStep and
  loadExecution. A failure that lands after the probe still stops
  planning, because createSteps refuses to insert after a failure.
- step:settled: skip refreshLiveStatus when it cannot change the status:
  after the handler finishes the execution, and after it queues steps for
  an execution that is running.
- v1 step data loader: load the execution and its steps in parallel.

Durability and ordering guarantees do not change. One visible effect:
if a branch suspends while another branch queues steps, the execution
can show 'waiting' instead of 'running' until the next settlement. The
engine treats both statuses as live, so only the reporting API sees it.

Webhook -> 3 NoOps: about 12 fewer sequential round trips on the
orchestrator and 3 fewer on the step worker per execution.

Co-authored-by: Cursor <cursoragent@cursor.com>

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

@@ -84,7 +84,14 @@ export class TypeOrmStepStore implements StepStore {
 
 		// Ids are assigned here because the entity's insert hook only runs on class
 		// instances, and these are plain values.
-		const rows = records.map((record) => ({ ...record, executionId, id: generateId() }));
+		const parameters: unknown[] = [executionId];
+		const tuples = records.map((record) => {
+			// jsonb is passed as text: the driver would send a JS array as a Postgres array.
+			const outputs = record.outputs === undefined ? null : JSON.stringify(record.outputs);
+			parameters.push(generateId(), record.nodeId, record.iteration, record.status, outputs);
+			const p = parameters.length;
+			return `($${p - 4}::uuid, $${p - 3}, $${p - 2}::int, $${p - 1}, $${p}::jsonb)`;
+		});
 
 		// The execution-row lock serializes this insert with `failStep`, so rows
 		// land before the failure (its sweep cancels them) or not at all.
@@ -93,31 +100,28 @@ export class TypeOrmStepStore implements StepStore {
 			await manager.query('SELECT id FROM workflow_execution WHERE id = $1 FOR SHARE', [
 				executionId,
 			]);
-			const [failed] = await manager.query<Array<{ id: string }>>(
-				`SELECT id FROM workflow_step_execution
-				 WHERE execution_id = $1 AND status = 'failed'
-				 LIMIT 1`,
-				[executionId],
+
+			// The failed-step test must be a statement after the lock: a statement
+			// that waits for the lock keeps the snapshot it started with, and would
+			// miss a failure committed during the wait.
+			//
+			// ON CONFLICT DO NOTHING is the unique key doing its job: a node another
+			// planner already queued is skipped, leaving the rest of the batch to land.
+			// RETURNING emits exactly the rows that were inserted.
+			const inserted = await manager.query<InsertedStepRow[]>(
+				`INSERT INTO workflow_step_execution (id, execution_id, node_id, iteration, status, outputs)
+				 SELECT candidate.id, $1::uuid, candidate.node_id, candidate.iteration, candidate.status, candidate.outputs
+				 FROM (VALUES ${tuples.join(', ')}) AS candidate (id, node_id, iteration, status, outputs)
+				 WHERE NOT EXISTS (
+					SELECT 1 FROM workflow_step_execution
+					WHERE execution_id = $1::uuid AND status = 'failed'
+				 )
+				 ON CONFLICT DO NOTHING
+				 RETURNING id, node_id, iteration`,
+				parameters,
 			);
-			if (failed) return [];
 
-			// `orIgnore` is the unique key doing its job: a node another planner already
-			// queued is skipped, leaving the rest of the batch to land. RETURNING emits
-			// exactly the rows that were inserted, keyed by database column name.
-			const result = await manager
-				.createQueryBuilder()
-				.insert()
-				.into(WorkflowStepExecution)
-				.values(rows)
-				.orIgnore()
-				.returning(['id', 'nodeId', 'iteration'])
-				.execute();
-
-			return (result.raw as InsertedStepRow[]).map(({ id, node_id: nodeId, iteration }) => ({
-				id,
-				nodeId,
-				iteration,
-			}));
+			return inserted.map(({ id, node_id: nodeId, iteration }) => ({ id, nodeId, iteration }));
 		});
 	}
 

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

@@ -481,13 +481,49 @@ describe('StepSettledHandler', () => {
 		expect(executionStore.finishExecution).not.toHaveBeenCalled();
 	});
 
-	it('reads the execution status off the steps again after every settlement', async () => {
+	it('reads the execution status off the steps again after a settlement that queues nothing', async () => {
+		const stepStore = makeStepStore(
+			{ id: 'step-m', nodeId: 'm' },
+			{ countSettledSteps: vi.fn().mockResolvedValue(4) },
+		);
+		const { handler, executionStore } = makeHandler(stepStore);
+
+		await handler.handle({ ...event, stepId: 'step-m' });
+
+		expect(executionStore.refreshLiveStatus).toHaveBeenCalledExactlyOnceWith('exec-1');
+	});
+
+	it('reads the execution status off the steps again after queueing work for a waiting execution', async () => {
+		const stepStore = makeStepStore();
+		const { handler, executionStore } = makeHandler(stepStore, {
+			executionStore: makeExecutionStore({ status: 'waiting' }),
+		});
+
+		await handler.handle(event);
+
+		expect(executionStore.refreshLiveStatus).toHaveBeenCalledExactlyOnceWith('exec-1');
+	});
+
+	it('does not refresh the status of a running execution after queueing work', async () => {
 		const stepStore = makeStepStore();
 		const { handler, executionStore } = makeHandler(stepStore);
 
 		await handler.handle(event);
 
-		expect(executionStore.refreshLiveStatus).toHaveBeenCalledExactlyOnceWith('exec-1');
+		expect(executionStore.refreshLiveStatus).not.toHaveBeenCalled();
+	});
+
+	it('does not refresh the status of an execution it just finished', async () => {
+		const stepStore = makeStepStore(
+			{ id: 'step-m', nodeId: 'm' },
+			{ countSettledSteps: vi.fn().mockResolvedValue(5) },
+		);
+		const { handler, executionStore } = makeHandler(stepStore);
+
+		await handler.handle({ ...event, stepId: 'step-m' });
+
+		expect(executionStore.finishExecution).toHaveBeenCalledWith('exec-1', 'completed');
+		expect(executionStore.refreshLiveStatus).not.toHaveBeenCalled();
 	});
 
 	it('plans successors for a waiting execution', async () => {

packages/@n8n/engine/src/execution/step-settled-handler.ts

@@ -46,9 +46,13 @@ export class StepSettledHandler {
 	) {}
 
 	async handle(event: StepSettledEvent): Promise<void> {
-		const [step, execution] = await Promise.all([
+		// The failure probe is read beside the loads, so it costs no round trip of
+		// its own. A failure that lands after it still stops planning, because
+		// `createSteps` refuses to insert once a step has failed.
+		const [step, execution, failedElsewhere] = await Promise.all([
 			this.stepStore.loadStep(event.stepId),
 			this.executionStore.loadExecution(event.executionId),
+			this.stepStore.hasFailedSteps(event.executionId),
 		]);
 		const node = validateStepContext(step, execution);
 
@@ -67,7 +71,7 @@ export class StepSettledHandler {
 		if (step.status === 'completed' || step.status === 'skipped') {
 			// a failure elsewhere may still have its settled event queued behind
 			// this one, so it must end the execution here, before planning
-			if (await this.stepStore.hasFailedSteps(execution.id)) {
+			if (failedElsewhere) {
 				await this.failExecution(execution, step, node);
 				return;
 			}
@@ -77,10 +81,15 @@ export class StepSettledHandler {
 
 		// If we've queued steps, we know the execution isn't done yet, so we
 		// definitely don't need to mark it finished.
-		if (queued === 0) await this.finishExecutionIfDone(execution, step, node);
+		if (queued === 0) {
+			// A finished execution is no longer live, so the refresh would write nothing.
+			if (await this.finishExecutionIfDone(execution, step, node)) return;
+		} else if (execution.status === 'running') {
+			// The queued rows keep it runnable, so the refresh would write nothing.
+			// The settlement of those rows refreshes again.
+			return;
+		}
 
-		// If this call just finished the execution, it is no longer live, and the
-		// refresh leaves it alone.
 		await this.executionStore.refreshLiveStatus(execution.id);
 	}
 
@@ -170,12 +179,14 @@ export class StepSettledHandler {
 	 * `(node, iteration)`, only exist for reachable nodes, and never unsettle, so
 	 * the count comparison cannot pass early — in-flight events and unplanned
 	 * successors both leave steps outstanding.
+	 *
+	 * Returns whether this call recorded the outcome.
 	 */
 	private async finishExecutionIfDone(
 		execution: ExecutionRecord,
 		step: StepRecord,
 		node: GraphNode,
-	): Promise<void> {
+	): Promise<boolean> {
 		const reachable = this.reachableNodeIds(execution);
 		const loops = deriveLoops(execution.graph);
 		const terminalIterations = await loadTerminalIterations(
@@ -190,10 +201,10 @@ export class StepSettledHandler {
 		const expected = countExpectedSettledSteps(loops, reachable, terminalIterations);
 		// A loop still running owes an unknown number of steps, so there is nothing
 		// to compare against yet.
-		if (expected === undefined) return;
+		if (expected === undefined) return false;
 
 		const settled = await this.stepStore.countSettledSteps(execution.id);
-		if (settled < expected) return;
+		if (settled < expected) return false;
 
 		const failed = await this.stepStore.hasFailedSteps(execution.id);
 		const finished = await this.executionStore.finishExecution(
@@ -209,6 +220,7 @@ export class StepSettledHandler {
 			});
 			this.announceEnd(execution, step, node, failed ? 'failed' : 'completed');
 		}
+		return finished !== null;
 	}
 
 	/**

packages/@n8n/node-engine-compatibility/src/engine-step-data-loader.ts

@@ -17,8 +17,10 @@ export function createEngineStepDataLoader(
 	stepStore: StepStore,
 ): StepDataLoader {
 	return async (context): Promise<StepData> => {
-		const execution = await executionStore.loadExecution(context.executionId);
-		const steps = await stepStore.loadAllSteps(context.executionId);
+		const [execution, steps] = await Promise.all([
+			executionStore.loadExecution(context.executionId),
+			stepStore.loadAllSteps(context.executionId),
+		]);
 
 		const byNode = new Map<string, Record<number, StepSlots>>();
 		for (const step of steps) {