diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts index 79f04e6a1f9..5e0fa696529 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts @@ -4,6 +4,7 @@ import postgresVersions from 'n8n-containers/postgres-versions.json'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { ExecutionNotFoundError } from '../../execution/execution-store'; +import type { ExecutionStatus, StepStatus } from '../../execution/execution.types'; import { createDataSource } from '../data-source'; import { WorkflowExecution } from '../entities/workflow-execution.entity'; import { WorkflowStepExecution } from '../entities/workflow-step-execution.entity'; @@ -177,4 +178,134 @@ describe('workflow_execution table (integration)', () => { expect(runningForWf2).toBe(1); }); + + it('TypeOrmExecutionStore.finishExecution ends a waiting execution', async () => { + const repo = dataSource.getRepository(WorkflowExecution); + const created = await repo.save( + repo.create({ + id: generateId(), + workflowId: 'wf-5', + status: 'waiting', + mode: 'production', + callerContext: { hostMode: 'trigger' }, + graph: { nodes: [], edges: [] }, + workflow: {}, + triggerOutputs: null, + finishedAt: null, + }), + ); + + const finished = await new TypeOrmExecutionStore(repo).finishExecution(created.id, 'failed'); + + expect(finished).toBe(true); + const row = await repo.findOneOrFail({ where: { id: created.id } }); + expect(row.status).toBe('failed'); + expect(row.finishedAt).toBeInstanceOf(Date); + }); + + it.each(['completed', 'queued'])( + 'TypeOrmExecutionStore.finishExecution leaves a %j execution alone', + async (status) => { + const repo = dataSource.getRepository(WorkflowExecution); + const finishedAt = status === 'completed' ? new Date('2099-01-01T00:00:00.000Z') : null; + const created = await repo.save( + repo.create({ + id: generateId(), + workflowId: 'wf-6', + status, + mode: 'production', + callerContext: { hostMode: 'trigger' }, + graph: { nodes: [], edges: [] }, + workflow: {}, + triggerOutputs: null, + finishedAt, + }), + ); + + const finished = await new TypeOrmExecutionStore(repo).finishExecution(created.id, 'failed'); + + expect(finished).toBe(false); + const row = await repo.findOneOrFail({ where: { id: created.id } }); + expect(row.status).toBe(status); + expect(row.finishedAt).toEqual(finishedAt); + }, + ); + + describe('TypeOrmExecutionStore.refreshLiveStatus', () => { + /** One execution with one step for each status given. */ + async function seed(status: ExecutionStatus, stepStatuses: StepStatus[]): Promise { + const repo = dataSource.getRepository(WorkflowExecution); + const execution = await repo.save( + repo.create({ + id: generateId(), + workflowId: 'wf-live', + status, + mode: 'production', + callerContext: { hostMode: 'trigger' }, + graph: { nodes: [], edges: [] }, + workflow: {}, + triggerOutputs: null, + finishedAt: null, + }), + ); + const stepRepo = dataSource.getRepository(WorkflowStepExecution); + await stepRepo.save( + // one iteration for each, so they are distinct rows of the same node + stepStatuses.map((stepStatus, iteration) => + stepRepo.create({ + id: generateId(), + executionId: execution.id, + nodeId: 'a', + iteration, + status: stepStatus, + outputs: null, + error: null, + waitDeclaration: null, + waitTill: null, + resumeCause: null, + }), + ), + ); + return execution.id; + } + + it.each<[string, ExecutionStatus, StepStatus[], ExecutionStatus]>([ + [ + 'reports waiting when every step it still owes is suspended', + 'running', + ['completed', 'waiting'], + 'waiting', + ], + ['reports running again once a step can run', 'waiting', ['waiting', 'queued'], 'running'], + ['reports running while a step runs', 'waiting', ['waiting', 'running'], 'running'], + [ + 'leaves an execution whose steps have all settled to finishExecution', + 'running', + ['completed'], + 'running', + ], + ['leaves an execution with no steps alone', 'running', [], 'running'], + ['leaves an execution that already ended alone', 'completed', ['waiting'], 'completed'], + ])('%s', async (_case, from, stepStatuses, expected) => { + const repo = dataSource.getRepository(WorkflowExecution); + const id = await seed(from, stepStatuses); + + await new TypeOrmExecutionStore(repo).refreshLiveStatus(id); + + const row = await repo.findOneOrFail({ where: { id } }); + expect(row.status).toBe(expected); + }); + + it('does not write when the status already holds', async () => { + const repo = dataSource.getRepository(WorkflowExecution); + const id = await seed('running', ['running']); + const before = await repo.findOneOrFail({ where: { id } }); + + await new TypeOrmExecutionStore(repo).refreshLiveStatus(id); + + // The timestamp is the only trace of the write: the status is the same. + const after = await repo.findOneOrFail({ where: { id } }); + expect(after.updatedAt).toEqual(before.updatedAt); + }); + }); }); diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts index f28aef952cc..cb48be207cf 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts @@ -1096,6 +1096,8 @@ describe('workflow_step_execution table (integration)', () => { expect(byName.idx_workflow_step_execution_wait_till).toContain( "WHERE ((status)::text = 'waiting'", ); + expect(byName.idx_workflow_step_execution_unsettled).toContain('(execution_id, status)'); + expect(byName.idx_workflow_step_execution_unsettled).toContain('WHERE ((status)::text = ANY'); const [column]: Array<{ is_nullable: string; column_default: string }> = await dataSource.query( `SELECT is_nullable, column_default FROM information_schema.columns diff --git a/packages/@n8n/engine/src/database/entities/workflow-step-execution.entity.ts b/packages/@n8n/engine/src/database/entities/workflow-step-execution.entity.ts index f01414cba20..393c5c28875 100644 --- a/packages/@n8n/engine/src/database/entities/workflow-step-execution.entity.ts +++ b/packages/@n8n/engine/src/database/entities/workflow-step-execution.entity.ts @@ -25,6 +25,9 @@ import { generateId } from '../generate-id'; ) @Index('idx_workflow_step_execution_failed', ['executionId'], { where: "status = 'failed'" }) @Index('idx_workflow_step_execution_wait_till', ['waitTill'], { where: "status = 'waiting'" }) +@Index('idx_workflow_step_execution_unsettled', ['executionId', 'status'], { + where: "status IN ('queued', 'running', 'waiting')", +}) export class WorkflowStepExecution { @PrimaryColumn('uuid') id!: string; diff --git a/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts b/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts index c0a8f1adc1f..571c01f77b8 100644 --- a/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts +++ b/packages/@n8n/engine/src/database/migrations/1778529600000-CreateWorkflowExecution.ts @@ -39,7 +39,8 @@ export class CreateWorkflowExecution1778529600000 implements MigrationInterface checks: [ { name: 'chk_workflow_execution_status', - expression: "status IN ('queued', 'running', 'completed', 'failed', 'cancelled')", + expression: + "status IN ('queued', 'running', 'waiting', 'completed', 'failed', 'cancelled')", }, ], }), diff --git a/packages/@n8n/engine/src/database/migrations/1784890100000-CreateWorkflowStepExecution.ts b/packages/@n8n/engine/src/database/migrations/1784890100000-CreateWorkflowStepExecution.ts index 82201f6756e..a49d4b1a94c 100644 --- a/packages/@n8n/engine/src/database/migrations/1784890100000-CreateWorkflowStepExecution.ts +++ b/packages/@n8n/engine/src/database/migrations/1784890100000-CreateWorkflowStepExecution.ts @@ -89,6 +89,13 @@ export class CreateWorkflowStepExecution1784890100000 implements MigrationInterf columnNames: ['wait_till'], where: "status = 'waiting'", }, + // For the live-status refresh, on every suspension and settlement. + // `status` in the key keeps its probes index-only; settled rows drop out. + { + name: 'idx_workflow_step_execution_unsettled', + columnNames: ['execution_id', 'status'], + where: "status IN ('queued', 'running', 'waiting')", + }, ], foreignKeys: [ { diff --git a/packages/@n8n/engine/src/database/typeorm-execution-store.ts b/packages/@n8n/engine/src/database/typeorm-execution-store.ts index f274cc68e27..db182dad7ee 100644 --- a/packages/@n8n/engine/src/database/typeorm-execution-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-execution-store.ts @@ -1,4 +1,4 @@ -import type { Repository } from '@n8n/typeorm'; +import { In, type Repository } from '@n8n/typeorm'; import type { WorkflowExecution } from './entities'; import { @@ -7,7 +7,7 @@ import { type ExecutionStore, type NewExecutionRecord, } from '../execution/execution-store'; -import type { ExecutionStatus } from '../execution/execution.types'; +import { LIVE_EXECUTION_STATUSES, type ExecutionStatus } from '../execution/execution.types'; /** * Insert payload accepted by the repository. Derived from the method rather than @@ -49,10 +49,42 @@ export class TypeOrmExecutionStore implements ExecutionStore { } async finishExecution(id: string, status: 'completed' | 'failed'): Promise { + // A waiting execution can end too: a failure elsewhere cancels its waits. const result = await this.repo.update( - { id, status: 'running' }, + { id, status: In([...LIVE_EXECUTION_STATUSES]) }, { status, finishedAt: new Date() }, ); return result.affected === 1; } + + async refreshLiveStatus(id: string): Promise { + // The probes read the steps before the UPDATE takes the row's lock, so a + // step can change in between. + await this.repo.query( + // MATERIALIZED, because `live.runnable` is read three times below. + `WITH live AS MATERIALIZED ( + SELECT + EXISTS ( + SELECT 1 FROM workflow_step_execution + WHERE execution_id = $1 AND status IN ('queued', 'running') + ) AS runnable, + EXISTS ( + SELECT 1 FROM workflow_step_execution + WHERE execution_id = $1 AND status = 'waiting' + ) AS waiting + ) + UPDATE workflow_execution e + SET status = CASE WHEN live.runnable THEN 'running' ELSE 'waiting' END, + -- A raw query bypasses the UpdateDateColumn hook, so set the time here. + updated_at = now() + FROM live + WHERE e.id = $1 + -- Re-checked against the row it locks, so losing to finishExecution writes nothing. + AND e.status IN ('running', 'waiting') + AND (live.runnable OR live.waiting) + -- Skip a write that changes nothing, so the row is not locked for it. + AND e.status IS DISTINCT FROM CASE WHEN live.runnable THEN 'running' ELSE 'waiting' END`, + [id], + ); + } } diff --git a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts index 72a97a0ed8e..efd41924d83 100644 --- a/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts @@ -28,6 +28,7 @@ function makeExecutionStore(overrides: Partial = {}): ExecutionS loadExecution: vi.fn(), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + refreshLiveStatus: vi.fn(), ...overrides, }; } diff --git a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts index beae1a88c4b..8d82a6dd91a 100644 --- a/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/start-execution.service.test.ts @@ -29,6 +29,7 @@ function makeStore(overrides: Partial = {}): ExecutionStore { loadExecution: vi.fn(), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + refreshLiveStatus: vi.fn(), ...overrides, }; } 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 74f59d7fc74..3da90ad22b3 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 @@ -237,7 +237,7 @@ describe('step execution (integration)', () => { expect(step?.outputs).toBeNull(); }); - it('suspends a step that declares a wait, leaving the execution running', async () => { + it('suspends a step that declares a wait, and reports the execution as waiting', async () => { const waitGraph: WorkflowGraph = { nodes: [ { id: 'trigger', name: 'Webhook', type: 'trigger' }, @@ -269,7 +269,7 @@ describe('step execution (integration)', () => { }); // nothing settled the step, so the execution has no outcome to record - expect(execution.status).toBe('running'); + expect(execution.status).toBe('waiting'); expect(execution.finishedAt).toBeNull(); const waiting = steps.find(({ nodeId }) => nodeId === 'node-a'); diff --git a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts index 8d88c472f52..1a532f4fc97 100644 --- a/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts @@ -88,6 +88,7 @@ function makeExecutionStore(overrides: Partial = {}): Execution loadExecution: vi.fn().mockResolvedValue(execution), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + refreshLiveStatus: vi.fn(), }; } @@ -523,6 +524,25 @@ describe('StepReadyHandler', () => { expect(queue.publish).not.toHaveBeenCalled(); }); + it('runs the step when the execution is waiting', async () => { + const stepStore = makeStepStore(); + const queue = makeQueue(); + const executor = makeExecutor(); + const handler = makeHandler(makeExecutionStore({ status: 'waiting' }), stepStore, queue, { + v1StepExecutor: executor, + }); + + await handler.handle(event); + + expect(executor.execute).toHaveBeenCalledOnce(); + expect(stepStore.completeStep).toHaveBeenCalledWith('step-a', [[{ json: { ok: true } }]]); + expect(queue.publish).toHaveBeenCalledExactlyOnceWith({ + type: 'step:settled', + executionId: 'exec-1', + stepId: 'step-a', + }); + }); + it('does not report completion when the lifecycle event is not recorded', async () => { const stepStore = makeStepStore({}, { completeStep: vi.fn().mockResolvedValue(false) }); const queue = makeQueue(); @@ -737,7 +757,8 @@ describe('StepReadyHandler waits', () => { it('suspends the step and announces no settlement when the executor declares a wait', async () => { const stepStore = makeStepStore(); const queue = makeQueue(); - const handler = makeHandler(makeExecutionStore(), stepStore, queue, { + const executionStore = makeExecutionStore(); + const handler = makeHandler(executionStore, stepStore, queue, { v1StepExecutor: makeExecutor({ wait: timeWait }), }); @@ -749,6 +770,7 @@ describe('StepReadyHandler waits', () => { expect(stepStore.completeStep).not.toHaveBeenCalled(); expect(stepStore.failStep).not.toHaveBeenCalled(); expect(queue.publish).not.toHaveBeenCalled(); + expect(executionStore.refreshLiveStatus).toHaveBeenCalledExactlyOnceWith('exec-1'); }); it('reports the suspension once the row is written, so the sweeper can re-arm', async () => { 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 0a1348bd9c7..2f7af6d0c57 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 @@ -65,6 +65,7 @@ function makeExecutionStore( loadExecution: vi.fn().mockResolvedValue(execution), transitionStatus: vi.fn().mockResolvedValue(true), finishExecution: vi.fn().mockResolvedValue(true), + refreshLiveStatus: vi.fn(), ...storeOverrides, }; } @@ -474,6 +475,30 @@ describe('StepSettledHandler', () => { expect(executionStore.finishExecution).not.toHaveBeenCalled(); }); + it('reads the execution status off the steps again after every settlement', async () => { + const stepStore = makeStepStore(); + const { handler, executionStore } = makeHandler(stepStore); + + await handler.handle(event); + + expect(executionStore.refreshLiveStatus).toHaveBeenCalledExactlyOnceWith('exec-1'); + }); + + it('plans successors for a waiting execution', async () => { + const stepStore = makeStepStore(); + const { handler, stepQueue } = makeHandler(stepStore, { + executionStore: makeExecutionStore({ status: 'waiting' }), + }); + + await handler.handle(event); + + expect(stepStore.createSteps).toHaveBeenCalledExactlyOnceWith('exec-1', [ + { nodeId: 'b', iteration: 0, status: 'queued' }, + { nodeId: 'c', iteration: 0, status: 'queued' }, + ]); + expect(stepQueue.publish).toHaveBeenCalledTimes(2); + }); + it('does not test for completion when it just queued work', async () => { const stepStore = makeStepStore(); const { handler, executionStore } = makeHandler(stepStore); diff --git a/packages/@n8n/engine/src/execution/execution-store.ts b/packages/@n8n/engine/src/execution/execution-store.ts index 535ac68c156..ece60a62be8 100644 --- a/packages/@n8n/engine/src/execution/execution-store.ts +++ b/packages/@n8n/engine/src/execution/execution-store.ts @@ -55,7 +55,21 @@ export interface ExecutionStore { /** * Record an execution's outcome: writes the final status and the finish time - * together, as a compare-and-set on `running`, so they can't be observed apart. + * together, as a compare-and-set on the live statuses, so they can't be + * observed apart. */ finishExecution(id: string, status: 'completed' | 'failed'): Promise; + + /** + * Sets a live execution's status from the state of its steps: `waiting` when + * every step it still owes is suspended, `running` when one step can run. + * + * Call this when a step suspends and when a step settles. A resume does not + * call it, so the status can lag until the resumed step settles. + * + * The refresh takes no lock, so it can lose a race and confuse `running` + * with `waiting`. Callers may test whether the execution ended, as both step + * handlers do. Read the step rows to tell `running` from `waiting`. + */ + refreshLiveStatus(id: string): Promise; } diff --git a/packages/@n8n/engine/src/execution/execution.types.ts b/packages/@n8n/engine/src/execution/execution.types.ts index ee5f4b2090b..a7afcc02dba 100644 --- a/packages/@n8n/engine/src/execution/execution.types.ts +++ b/packages/@n8n/engine/src/execution/execution.types.ts @@ -1,7 +1,27 @@ import type { JsonObject, JsonValue } from '../common'; -/** Lifecycle status of an execution. */ -export type ExecutionStatus = 'queued' | 'running' | 'completed' | 'failed' | 'cancelled'; +/** + * Lifecycle status of an execution. A resume does not update it, so it can + * read `waiting` while the resumed step runs. See + * `ExecutionStore.refreshLiveStatus`. + */ +export type ExecutionStatus = + | 'queued' + | 'running' + | 'waiting' + | 'completed' + | 'failed' + | 'cancelled'; + +/** + * Started and not ended. `queued` is not live: it has no step rows yet. The SQL + * in `TypeOrmExecutionStore.refreshLiveStatus` repeats this list. + */ +export const LIVE_EXECUTION_STATUSES = ['running', 'waiting'] as const; + +export function isLiveExecutionStatus(status: ExecutionStatus): boolean { + return (LIVE_EXECUTION_STATUSES as readonly ExecutionStatus[]).includes(status); +} /** How an execution was initiated. */ export type ExecutionMode = 'production' | 'manual'; diff --git a/packages/@n8n/engine/src/execution/step-ready-handler.ts b/packages/@n8n/engine/src/execution/step-ready-handler.ts index 5c04b90837a..156baf37ce3 100644 --- a/packages/@n8n/engine/src/execution/step-ready-handler.ts +++ b/packages/@n8n/engine/src/execution/step-ready-handler.ts @@ -14,6 +14,7 @@ import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; import { stepKeyId, + isLiveExecutionStatus, isSettledStatus, hasResumeCondition, type ResumeCause, @@ -75,10 +76,10 @@ export class StepReadyHandler { const executor = node.type === 'batch' || step.resumeCause !== null ? undefined : this.executorFor(step, node); - if (execution.status !== 'running') { - // The execution is no longer running, so we don't run the step. - // The step is left `running` for reconciliation (CAT-2938) or - // internal consistency checks (CAT-3930) to resolve. + if (!isLiveExecutionStatus(execution.status)) { + // The execution has ended, so we don't run the step. The step is left + // `running` for reconciliation (CAT-2938) or internal consistency + // checks (CAT-3930) to resolve. return; } @@ -137,9 +138,13 @@ export class StepReadyHandler { if (!recorded) return; // A wait is no outcome: nothing settled, so nothing is announced and no - // planning follows. TODO(CAT-2928): publish `step:waiting` so the UI can show it. + // planning follows. + // TODO(CAT-2928): publish `step:waiting` so the UI can show it. if (run.kind === 'wait') { + // Nudge first, so the sweeper's re-arm read runs beside the status + // write rather than after it. Either order is correct. this.onStepSuspended(); + await this.executionStore.refreshLiveStatus(execution.id); return; } diff --git a/packages/@n8n/engine/src/execution/step-settled-handler.ts b/packages/@n8n/engine/src/execution/step-settled-handler.ts index 8351877b425..303fee67605 100644 --- a/packages/@n8n/engine/src/execution/step-settled-handler.ts +++ b/packages/@n8n/engine/src/execution/step-settled-handler.ts @@ -11,7 +11,13 @@ import type { ExecutionResponseSender } from '../response-channel'; import type { OrchestrationMessage, StepMessage, StepSettledEvent, WorkQueue } from '../queue'; import { countExpectedSettledSteps } from './completion'; import type { ExecutionRecord, ExecutionStore } from './execution-store'; -import { isSettledStatus, stepKeyId, type StepKey, type StepKeyId } from './execution.types'; +import { + isLiveExecutionStatus, + isSettledStatus, + stepKeyId, + type StepKey, + type StepKeyId, +} from './execution.types'; import { exitSourcesInto, loadTerminalIterations } from './loop-ledger'; import { decideSuccessors, decisionKeys } from './settlement'; import type { StepRecord, StepStore } from './step-store'; @@ -53,7 +59,9 @@ export class StepSettledHandler { return; } - if (execution.status !== 'running') return; + // A `waiting` execution is live, and this settlement may be what lets it + // move on, so only an ended one stops here. + if (!isLiveExecutionStatus(execution.status)) return; let queued = 0; if (step.status === 'completed' || step.status === 'skipped') { @@ -69,9 +77,11 @@ 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) return; + if (queued === 0) await this.finishExecutionIfDone(execution, step, node); - await this.finishExecutionIfDone(execution, step, node); + // If this call just finished the execution, it is no longer live, and the + // refresh leaves it alone. + await this.executionStore.refreshLiveStatus(execution.id); } private async failExecution( 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 2318bf4a0d8..d7a5a8f7a6f 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 @@ -72,6 +72,11 @@ describe('POST /api/workflow-executions/search (integration)', () => { await search(body).expect(400); }); + // An execution reports `waiting`, so the control plane filters the list by it. + it('accepts a search for waiting executions', async () => { + await search({ workflowIds: 'all', status: ['waiting'] }).expect(200); + }); + it('rejects a cursor paired with a status-first sort at the store', async () => { const { executionViewStore } = createStores(dataSource); await expect( diff --git a/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts b/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts index 7cdb4ed14f2..d5dffd8271a 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts @@ -22,7 +22,14 @@ const GetExecutionQuery = z.object({ includeSteps: z.enum(['true', 'false']).opt const datetimeStringWithOffset = () => z.string().datetime({ offset: true }); -const ExecutionStatusSchema = z.enum(['queued', 'running', 'completed', 'failed', 'cancelled']); +const ExecutionStatusSchema = z.enum([ + 'queued', + 'running', + 'waiting', + 'completed', + 'failed', + 'cancelled', +]); const SearchExecutionsBody = z .object({ diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts index 5b92fd1b589..335c1ada92c 100644 --- a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts @@ -57,15 +57,13 @@ describe('toV1RunExecutionData', () => { ['completed', 'success'], ['failed', 'error'], ['running', 'running'], + ['waiting', 'waiting'], ])('maps step status %j to %j', (status, expected) => { const data = toV1RunExecutionData(graph, [step({ status })]); expect(data.resultData.runData.Trigger[0].executionStatus).toBe(expected); }); - // A cancelled step reports no run. `cancelPendingSteps` cancels a waiting step - // too, and that step did run, so this case also covers a run that the map - // hides. The status mapping work decides what the canvas shows for it. it.each(['queued', 'skipped', 'cancelled'])( 'reports no run for a %j step, the way v1 reports a node that did not run', (status) => { @@ -75,6 +73,25 @@ describe('toV1RunExecutionData', () => { }, ); + it('reports a suspended step, and names it as the last node executed', async () => { + const data = toV1RunExecutionData(graph, [ + step(), + step({ + id: 'step-2', + nodeId: 's', + status: 'waiting', + outputs: null, + updatedAt: '2026-08-25T10:00:01.000Z', + }), + ]); + + expect(data.resultData.runData.Set[0].executionStatus).toBe('waiting'); + // v1 carries the node's pass-through data here. The step has none until it + // resumes, and the read path is not given the declaration that holds it. + expect(data.resultData.runData.Set[0].data).toBeUndefined(); + expect(data.resultData.lastNodeExecuted).toBe('Set'); + }); + it('takes timing from the row timestamps', () => { const data = toV1RunExecutionData(graph, [step()]); diff --git a/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts b/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts index d672d1be991..d0ca4795b32 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts @@ -32,15 +32,24 @@ import { emptyRun, forwardEdgesByTarget, nodeNamesById, toSourceSlots } from './ * having no `runData` for it, and an entry would make the canvas claim it ran. * `queued` never started and `skipped` was decided against. * - * Two statuses do not obey that rule. A `waiting` step ran and then suspended. - * A `cancelled` step can be a waiting step that `cancelPendingSteps` cancelled. - * Neither gets an entry today, so this map hides a run that happened. - * TODO(CAT-2928): map the waiting statuses the canvas must show. + * A `waiting` step gets an entry, because it did run: it started, and then it + * suspended. Engine v1 does the same. The run data of a waiting v1 execution + * holds a task for the paused node, with the status `waiting`. + * + * That entry carries no outputs. The step produces none until it resumes. + * Engine v1 puts the node's pass-through data there instead. This read cannot: + * those slots live in the step's wait declaration, which `StepDetail` does not + * report. + * + * TODO(CAT-2928): a `cancelled` step gets no entry. That is correct for a step + * that never started, but `cancelPendingSteps` also cancels the waiting steps + * when an execution ends, and those steps did run. The canvas hides them. */ const TASK_STATUS_V1 = new Map([ ['completed', 'success'], ['failed', 'error'], ['running', 'running'], + ['waiting', 'waiting'], ]); interface StepRun { diff --git a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts index 41c25b3b3a7..ed0ca981f77 100644 --- a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts +++ b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts @@ -75,10 +75,11 @@ describe('EngineV2ExecutionReader', () => { it('maps supported statuses and drops statuses that the DP cannot match', async () => { await reader.findMany({ ...query, status: ['success', 'crashed', 'waiting'] }, 'all'); expect(dataPlane.searchExecutions).toHaveBeenCalledWith( - expect.objectContaining({ status: ['completed'] }), + expect.objectContaining({ status: ['waiting', 'completed'] }), ); dataPlane.searchExecutions.mockClear(); - await reader.findMany({ ...query, status: ['crashed', 'waiting'] }, 'all'); + // `crashed` has no v2 counterpart, so nothing is left to search for + await reader.findMany({ ...query, status: ['crashed'] }, 'all'); expect(dataPlane.searchExecutions).not.toHaveBeenCalled(); }); @@ -226,6 +227,7 @@ describe('EngineV2ExecutionReader', () => { it.each<[ExecutionStatus, ExecutionStatusV1, boolean]>([ ['queued', 'new', false], ['running', 'running', false], + ['waiting', 'waiting', false], ['completed', 'success', true], ['failed', 'error', false], ['cancelled', 'canceled', false], diff --git a/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts b/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts index c21f31f2e9e..0de73851484 100644 --- a/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts +++ b/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts @@ -219,8 +219,11 @@ describe('ExecutionListV2Service', () => { expect(searchRequests()).toEqual( expect.arrayContaining([ + // `waiting` belongs to the completed block, not the current one, because + // v1 counts only `new` and `running` as current. The two engines must + // put a waiting execution in the same block. expect.objectContaining({ - status: ['completed', 'failed', 'cancelled'], + status: ['waiting', 'completed', 'failed', 'cancelled'], includeTotal: true, }), expect.objectContaining({ diff --git a/packages/cli/src/executions/engine-v2-mapping.ts b/packages/cli/src/executions/engine-v2-mapping.ts index 8fe6b07a4e5..fd96791fa17 100644 --- a/packages/cli/src/executions/engine-v2-mapping.ts +++ b/packages/cli/src/executions/engine-v2-mapping.ts @@ -10,6 +10,7 @@ import { const V1_STATUS_BY_V2_STATUS = new Map([ ['queued', 'new'], ['running', 'running'], + ['waiting', 'waiting'], ['completed', 'success'], ['failed', 'error'], ['cancelled', 'canceled'],