mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat(engine): Store the execution's waiting status (no-changelog) (#39093)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
a9247247af
commit
c53a5e4dd8
|
|
@ -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<ExecutionStatus>(['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<string> {
|
||||
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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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')",
|
||||
},
|
||||
],
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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: [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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<boolean> {
|
||||
// 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<void> {
|
||||
// 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],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ function makeExecutionStore(overrides: Partial<ExecutionStore> = {}): ExecutionS
|
|||
loadExecution: vi.fn(),
|
||||
transitionStatus: vi.fn().mockResolvedValue(true),
|
||||
finishExecution: vi.fn().mockResolvedValue(true),
|
||||
refreshLiveStatus: vi.fn(),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ function makeStore(overrides: Partial<ExecutionStore> = {}): ExecutionStore {
|
|||
loadExecution: vi.fn(),
|
||||
transitionStatus: vi.fn().mockResolvedValue(true),
|
||||
finishExecution: vi.fn().mockResolvedValue(true),
|
||||
refreshLiveStatus: vi.fn(),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -88,6 +88,7 @@ function makeExecutionStore(overrides: Partial<ExecutionRecord> = {}): 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 () => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<boolean>;
|
||||
|
||||
/**
|
||||
* 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<void>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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<StepStatus>(['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()]);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<StepStatus, ExecutionStatus>([
|
||||
['completed', 'success'],
|
||||
['failed', 'error'],
|
||||
['running', 'running'],
|
||||
['waiting', 'waiting'],
|
||||
]);
|
||||
|
||||
interface StepRun {
|
||||
|
|
|
|||
|
|
@ -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],
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import {
|
|||
const V1_STATUS_BY_V2_STATUS = new Map<ExecutionStatus, ExecutionStatusV1>([
|
||||
['queued', 'new'],
|
||||
['running', 'running'],
|
||||
['waiting', 'waiting'],
|
||||
['completed', 'success'],
|
||||
['failed', 'error'],
|
||||
['cancelled', 'canceled'],
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user