diff --git a/packages/@n8n/engine/src/database/typeorm-step-store.ts b/packages/@n8n/engine/src/database/typeorm-step-store.ts index ed0fa1d8d32..d3aff1cbf0c 100644 --- a/packages/@n8n/engine/src/database/typeorm-step-store.ts +++ b/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>( - `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( + `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 })); }); } diff --git a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts index f0f6337ea9b..bc1bfb33700 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts +++ b/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 () => { diff --git a/packages/@n8n/engine/src/execution/step-settled-handler.ts b/packages/@n8n/engine/src/execution/step-settled-handler.ts index 35c8c4482fb..22610de8fe4 100644 --- a/packages/@n8n/engine/src/execution/step-settled-handler.ts +++ b/packages/@n8n/engine/src/execution/step-settled-handler.ts @@ -46,9 +46,13 @@ export class StepSettledHandler { ) {} async handle(event: StepSettledEvent): Promise { - 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 { + ): Promise { 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; } /** diff --git a/packages/@n8n/node-engine-compatibility/src/engine-step-data-loader.ts b/packages/@n8n/node-engine-compatibility/src/engine-step-data-loader.ts index 56ac4373dcd..bc5a8636bfb 100644 --- a/packages/@n8n/node-engine-compatibility/src/engine-step-data-loader.ts +++ b/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 => { - 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>(); for (const step of steps) {