mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-11 22:50:06 +00:00
feat(engine): Return run data for engine 2.0 executions (#37486)
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5
parent
4a53faff09
commit
ae26d9599d
@@ -87,16 +87,19 @@ describe('workflow_execution table (integration)', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('TypeOrmExecutionViewStore.loadExecutionView throws for an unknown id', async () => {
|
||||
const viewStore = new TypeOrmExecutionViewStore(
|
||||
dataSource.getRepository(WorkflowExecution),
|
||||
dataSource.getRepository(WorkflowStepExecution),
|
||||
);
|
||||
it.each(['loadExecutionView', 'loadExecutionWithStepsView'] as const)(
|
||||
'TypeOrmExecutionViewStore.%s throws for an unknown id',
|
||||
async (method) => {
|
||||
const viewStore = new TypeOrmExecutionViewStore(
|
||||
dataSource.getRepository(WorkflowExecution),
|
||||
dataSource.getRepository(WorkflowStepExecution),
|
||||
);
|
||||
|
||||
await expect(
|
||||
viewStore.loadExecutionView('00000000-0000-0000-0000-000000000000'),
|
||||
).rejects.toBeInstanceOf(ExecutionNotFoundError);
|
||||
});
|
||||
await expect(
|
||||
viewStore[method]('00000000-0000-0000-0000-000000000000'),
|
||||
).rejects.toBeInstanceOf(ExecutionNotFoundError);
|
||||
},
|
||||
);
|
||||
|
||||
it('counts rows by workflowId and status (admittance support)', async () => {
|
||||
const repo = dataSource.getRepository(WorkflowExecution);
|
||||
|
||||
+11
-6
@@ -733,7 +733,7 @@ describe('workflow_step_execution table (integration)', () => {
|
||||
expect(column).toEqual({ is_nullable: 'NO', column_default: '0' });
|
||||
});
|
||||
|
||||
it('TypeOrmExecutionViewStore.loadStepViews returns every row of the execution, none of another', async () => {
|
||||
it('TypeOrmExecutionViewStore.loadExecutionWithStepsView returns every row of the execution, none of another', async () => {
|
||||
const executionId = await createExecution();
|
||||
const otherExecutionId = await createExecution();
|
||||
const store = new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution));
|
||||
@@ -743,18 +743,23 @@ describe('workflow_step_execution table (integration)', () => {
|
||||
]);
|
||||
await store.createSteps(otherExecutionId, [{ nodeId: 'c', iteration: 0, status: 'queued' }]);
|
||||
|
||||
const steps = await viewStore().loadStepViews(executionId);
|
||||
const { id, steps } = await viewStore().loadExecutionWithStepsView(executionId);
|
||||
|
||||
expect(id).toBe(executionId);
|
||||
expect(steps.map((step) => step.nodeId).sort()).toEqual(['a', 'b']);
|
||||
});
|
||||
|
||||
it('TypeOrmExecutionViewStore.loadStepViews returns [] for an execution with no steps', async () => {
|
||||
// The left join still has to report the execution itself.
|
||||
it('TypeOrmExecutionViewStore.loadExecutionWithStepsView returns [] for an execution with no steps', async () => {
|
||||
const executionId = await createExecution();
|
||||
|
||||
expect(await viewStore().loadStepViews(executionId)).toEqual([]);
|
||||
const view = await viewStore().loadExecutionWithStepsView(executionId);
|
||||
|
||||
expect(view.id).toBe(executionId);
|
||||
expect(view.steps).toEqual([]);
|
||||
});
|
||||
|
||||
it('TypeOrmExecutionViewStore.loadStepViews carries outputs, error and timestamps', async () => {
|
||||
it('TypeOrmExecutionViewStore.loadExecutionWithStepsView carries outputs, error and timestamps', async () => {
|
||||
const executionId = await createExecution();
|
||||
const store = new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution));
|
||||
await store.createSteps(executionId, [
|
||||
@@ -766,7 +771,7 @@ describe('workflow_step_execution table (integration)', () => {
|
||||
await store.claimStep(failedId);
|
||||
await store.failStep(failedId, { name: 'Error', message: 'boom' });
|
||||
|
||||
const steps = await viewStore().loadStepViews(executionId);
|
||||
const { steps } = await viewStore().loadExecutionWithStepsView(executionId);
|
||||
|
||||
const completed = steps.find((step) => step.nodeId === 'a');
|
||||
expect(completed).toMatchObject({
|
||||
|
||||
@@ -1,13 +1,21 @@
|
||||
import type { Repository } from '@n8n/typeorm';
|
||||
import type { Repository, SelectQueryBuilder } from '@n8n/typeorm';
|
||||
|
||||
import type { WorkflowExecution, WorkflowStepExecution } from './entities';
|
||||
import { ExecutionNotFoundError } from '../execution/execution-store';
|
||||
import type {
|
||||
ExecutionViewStore,
|
||||
ExecutionView,
|
||||
ExecutionWithStepsView,
|
||||
StepView,
|
||||
} from '../execution/execution-view-store';
|
||||
|
||||
type StepColumns = { [K in keyof StepView as `step_${K}`]: StepView[K] };
|
||||
/** What the left join yields for an execution that has no steps. */
|
||||
type NoStepColumns = { [K in keyof StepView as `step_${K}`]: null };
|
||||
|
||||
/** One joined row: the execution, plus one step of it or nothing. */
|
||||
type ExecutionStepRow = ExecutionView & (StepColumns | NoStepColumns);
|
||||
|
||||
/**
|
||||
* TypeORM-backed `ExecutionViewStore` adapter. It spans both tables, since a
|
||||
* read of an execution and a read of its steps are one concern.
|
||||
@@ -23,7 +31,38 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore {
|
||||
) {}
|
||||
|
||||
async loadExecutionView(id: string): Promise<ExecutionView> {
|
||||
const row: ExecutionView | undefined = await this.executions
|
||||
const row: ExecutionView | undefined = await this.selectExecution(id).getRawOne();
|
||||
if (!row) throw new ExecutionNotFoundError(id);
|
||||
return row;
|
||||
}
|
||||
|
||||
/**
|
||||
* A left join, so an execution with no steps still returns its own row. The
|
||||
* execution columns repeat for each step, `graph` included; that is the price
|
||||
* of reading the status and the steps as of one point in time.
|
||||
*/
|
||||
async loadExecutionWithStepsView(id: string): Promise<ExecutionWithStepsView> {
|
||||
const rows: ExecutionStepRow[] = await this.selectExecution(id)
|
||||
.addSelect('step.id', 'step_id')
|
||||
.addSelect('step.node_id', 'step_nodeId')
|
||||
.addSelect('step.iteration', 'step_iteration')
|
||||
.addSelect('step.status', 'step_status')
|
||||
.addSelect('step.outputs', 'step_outputs')
|
||||
.addSelect('step.error', 'step_error')
|
||||
.addSelect('step.created_at', 'step_createdAt')
|
||||
.addSelect('step.updated_at', 'step_updatedAt')
|
||||
.leftJoin(this.steps.metadata.tableName, 'step', 'step.execution_id = execution.id')
|
||||
.orderBy('step.created_at', 'ASC')
|
||||
.addOrderBy('step.node_id', 'ASC')
|
||||
.addOrderBy('step.iteration', 'ASC')
|
||||
.getRawMany();
|
||||
if (rows.length === 0) throw new ExecutionNotFoundError(id);
|
||||
|
||||
return { ...toExecutionView(rows[0]), steps: rows.filter(hasStep).map(toStepView) };
|
||||
}
|
||||
|
||||
private selectExecution(id: string): SelectQueryBuilder<WorkflowExecution> {
|
||||
return this.executions
|
||||
.createQueryBuilder('execution')
|
||||
.select('execution.id', 'id')
|
||||
.addSelect('execution.workflow_id', 'workflowId')
|
||||
@@ -33,28 +72,28 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore {
|
||||
.addSelect('execution.created_at', 'createdAt')
|
||||
.addSelect('execution.updated_at', 'updatedAt')
|
||||
.addSelect('execution.finished_at', 'finishedAt')
|
||||
.where('execution.id = :id', { id })
|
||||
.getRawOne();
|
||||
if (!row) throw new ExecutionNotFoundError(id);
|
||||
return row;
|
||||
}
|
||||
|
||||
async loadStepViews(executionId: string): Promise<StepView[]> {
|
||||
const rows: StepView[] = await this.steps
|
||||
.createQueryBuilder('step')
|
||||
.select('step.id', 'id')
|
||||
.addSelect('step.node_id', 'nodeId')
|
||||
.addSelect('step.iteration', 'iteration')
|
||||
.addSelect('step.status', 'status')
|
||||
.addSelect('step.outputs', 'outputs')
|
||||
.addSelect('step.error', 'error')
|
||||
.addSelect('step.created_at', 'createdAt')
|
||||
.addSelect('step.updated_at', 'updatedAt')
|
||||
.where('step.execution_id = :executionId', { executionId })
|
||||
.orderBy('step.created_at', 'ASC')
|
||||
.addOrderBy('step.node_id', 'ASC')
|
||||
.addOrderBy('step.iteration', 'ASC')
|
||||
.getRawMany();
|
||||
return rows;
|
||||
.where('execution.id = :id', { id });
|
||||
}
|
||||
}
|
||||
|
||||
function hasStep(row: ExecutionStepRow): row is ExecutionView & StepColumns {
|
||||
return row.step_id !== null;
|
||||
}
|
||||
|
||||
function toExecutionView(row: ExecutionStepRow): ExecutionView {
|
||||
const { id, workflowId, status, mode, graph, createdAt, updatedAt, finishedAt } = row;
|
||||
return { id, workflowId, status, mode, graph, createdAt, updatedAt, finishedAt };
|
||||
}
|
||||
|
||||
function toStepView(row: StepColumns): StepView {
|
||||
return {
|
||||
id: row.step_id,
|
||||
nodeId: row.step_nodeId,
|
||||
iteration: row.step_iteration,
|
||||
status: row.step_status,
|
||||
outputs: row.step_outputs,
|
||||
error: row.step_error,
|
||||
createdAt: row.step_createdAt,
|
||||
updatedAt: row.step_updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -2,12 +2,17 @@ import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { ExecutionQueryService } from '../execution-query.service';
|
||||
import { ExecutionNotFoundError } from '../execution-store';
|
||||
import type { ExecutionViewStore, ExecutionView, StepView } from '../execution-view-store';
|
||||
import type {
|
||||
ExecutionViewStore,
|
||||
ExecutionView,
|
||||
ExecutionWithStepsView,
|
||||
StepView,
|
||||
} from '../execution-view-store';
|
||||
|
||||
function makeViewStore(overrides: Partial<ExecutionViewStore> = {}): ExecutionViewStore {
|
||||
return {
|
||||
loadExecutionView: vi.fn(),
|
||||
loadStepViews: vi.fn(),
|
||||
loadExecutionWithStepsView: vi.fn(),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
@@ -33,12 +38,29 @@ describe('ExecutionQueryService', () => {
|
||||
await expect(service.getExecution('exec-1')).rejects.toBeInstanceOf(ExecutionNotFoundError);
|
||||
});
|
||||
|
||||
it('getSteps loads and returns the step views', async () => {
|
||||
const steps = [{ id: 'step-1' }] as StepView[];
|
||||
const viewStore = makeViewStore({ loadStepViews: vi.fn().mockResolvedValue(steps) });
|
||||
it('getExecutionWithSteps loads the execution and its steps in one call', async () => {
|
||||
const execution = {
|
||||
id: 'exec-1',
|
||||
steps: [{ id: 'step-1' }] as StepView[],
|
||||
} as ExecutionWithStepsView;
|
||||
const viewStore = makeViewStore({
|
||||
loadExecutionWithStepsView: vi.fn().mockResolvedValue(execution),
|
||||
});
|
||||
const service = new ExecutionQueryService(viewStore);
|
||||
|
||||
await expect(service.getSteps('exec-1')).resolves.toBe(steps);
|
||||
expect(viewStore.loadStepViews).toHaveBeenCalledWith('exec-1');
|
||||
await expect(service.getExecutionWithSteps('exec-1')).resolves.toBe(execution);
|
||||
expect(viewStore.loadExecutionWithStepsView).toHaveBeenCalledWith('exec-1');
|
||||
expect(viewStore.loadExecutionView).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('getExecutionWithSteps propagates ExecutionNotFoundError', async () => {
|
||||
const viewStore = makeViewStore({
|
||||
loadExecutionWithStepsView: vi.fn().mockRejectedValue(new ExecutionNotFoundError('exec-1')),
|
||||
});
|
||||
const service = new ExecutionQueryService(viewStore);
|
||||
|
||||
await expect(service.getExecutionWithSteps('exec-1')).rejects.toBeInstanceOf(
|
||||
ExecutionNotFoundError,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
import type { ExecutionViewStore, ExecutionView, StepView } from './execution-view-store';
|
||||
import type {
|
||||
ExecutionViewStore,
|
||||
ExecutionView,
|
||||
ExecutionWithStepsView,
|
||||
} from './execution-view-store';
|
||||
|
||||
/**
|
||||
* Read path for executions and their steps: the seam between the HTTP layer
|
||||
@@ -13,8 +17,13 @@ export class ExecutionQueryService {
|
||||
return await this.viewStore.loadExecutionView(id);
|
||||
}
|
||||
|
||||
/** Every step of the execution, oldest first. `[]` if it has none yet. */
|
||||
async getSteps(id: string): Promise<StepView[]> {
|
||||
return await this.viewStore.loadStepViews(id);
|
||||
/**
|
||||
* The execution and every step of it, oldest first. One query, so a settle
|
||||
* between two reads cannot report a status that predates the steps.
|
||||
*
|
||||
* @throws {ExecutionNotFoundError} if absent.
|
||||
*/
|
||||
async getExecutionWithSteps(id: string): Promise<ExecutionWithStepsView> {
|
||||
return await this.viewStore.loadExecutionWithStepsView(id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,6 +42,12 @@ export interface StepView {
|
||||
updatedAt: Date;
|
||||
}
|
||||
|
||||
/** An execution with its steps, read as one query so the two cannot disagree. */
|
||||
export interface ExecutionWithStepsView extends ExecutionView {
|
||||
/** Every step of the execution, oldest first. Empty if it has run none yet. */
|
||||
steps: StepView[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Persistence interface for the read path, held apart from `ExecutionStore` and
|
||||
* `StepStore` so a reader cannot reach a transition, and so read concerns
|
||||
@@ -56,6 +62,10 @@ export interface ExecutionViewStore {
|
||||
/** Read view of one execution. Throws `ExecutionNotFoundError` if absent. */
|
||||
loadExecutionView(id: string): Promise<ExecutionView>;
|
||||
|
||||
/** Read views of every step of the execution, oldest first. Empty if it has none yet. */
|
||||
loadStepViews(executionId: string): Promise<StepView[]>;
|
||||
/**
|
||||
* Read view of one execution and its steps. One query, so the status a caller
|
||||
* reports cannot predate the steps it reports beside it. Throws
|
||||
* `ExecutionNotFoundError` if absent.
|
||||
*/
|
||||
loadExecutionWithStepsView(id: string): Promise<ExecutionWithStepsView>;
|
||||
}
|
||||
|
||||
@@ -16,7 +16,12 @@ export type {
|
||||
} from './execution.types';
|
||||
export { ExecutionNotFoundError } from './execution-store';
|
||||
export type { ExecutionRecord, ExecutionStore, NewExecutionRecord } from './execution-store';
|
||||
export type { ExecutionViewStore, ExecutionView, StepView } from './execution-view-store';
|
||||
export type {
|
||||
ExecutionViewStore,
|
||||
ExecutionView,
|
||||
ExecutionWithStepsView,
|
||||
StepView,
|
||||
} from './execution-view-store';
|
||||
export { StepNotFoundError } from './step-store';
|
||||
export type { NewStepRecord, StepRecord, StepStore, StepSummary } from './step-store';
|
||||
export { ExecutionStartHandler } from './execution-start-handler';
|
||||
|
||||
@@ -15,12 +15,7 @@ export {
|
||||
} from './auth';
|
||||
export type { AuthenticatedCaller, ActionScope, IdentityVerifier } from './auth';
|
||||
|
||||
export type {
|
||||
EngineErrorResponse,
|
||||
ExecutionSnapshot,
|
||||
ExecutionStepsResponse,
|
||||
StepDetail,
|
||||
} from './server';
|
||||
export type { EngineErrorResponse, ExecutionSnapshot, StepDetail } from './server';
|
||||
|
||||
// The publisher stays internal: no host constructs or swaps one.
|
||||
export {
|
||||
|
||||
@@ -252,6 +252,7 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
|
||||
graph: sampleGraph,
|
||||
finishedAt: null,
|
||||
});
|
||||
expect(response.body).not.toHaveProperty('steps');
|
||||
const body = response.body as { createdAt: string; updatedAt: string };
|
||||
expect(new Date(body.createdAt).toISOString()).toBe(body.createdAt);
|
||||
expect(new Date(body.updatedAt).toISOString()).toBe(body.updatedAt);
|
||||
@@ -304,7 +305,8 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
|
||||
await finished;
|
||||
|
||||
const response = await request(runtime.app)
|
||||
.get(`/api/workflow-executions/${executionId}/steps`)
|
||||
.get(`/api/workflow-executions/${executionId}`)
|
||||
.query({ includeSteps: 'true' })
|
||||
.set(authHeader());
|
||||
await runtime.stop();
|
||||
|
||||
@@ -320,18 +322,48 @@ describe('GET /api/workflow-executions/:id (integration)', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('returns 200 with an empty list from /steps for an unknown execution id', async () => {
|
||||
it('returns an empty step list for an execution that has run nothing yet', async () => {
|
||||
const executionId = await createExecution();
|
||||
|
||||
const response = await request(url)
|
||||
.get('/api/workflow-executions/00000000-0000-0000-0000-000000000000/steps')
|
||||
.get(`/api/workflow-executions/${executionId}`)
|
||||
.query({ includeSteps: 'true' })
|
||||
.set(authHeader());
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body).toEqual({ steps: [] });
|
||||
expect((response.body as { steps: unknown[] }).steps).toEqual([]);
|
||||
});
|
||||
|
||||
it('returns 400 invalid_request from /steps for a non-uuid id', async () => {
|
||||
it('returns 404 not_found for an unknown execution id even when steps are asked for', async () => {
|
||||
const response = await request(url)
|
||||
.get('/api/workflow-executions/not-a-uuid/steps')
|
||||
.get('/api/workflow-executions/00000000-0000-0000-0000-000000000000')
|
||||
.query({ includeSteps: 'true' })
|
||||
.set(authHeader());
|
||||
|
||||
expect(response.status).toBe(404);
|
||||
expect((response.body as { error: string }).error).toBe('not_found');
|
||||
});
|
||||
|
||||
it('returns 400 invalid_request for a misspelled query key', async () => {
|
||||
const executionId = await createExecution();
|
||||
|
||||
// Stripping it would answer 200 with no steps, which reads the same as an
|
||||
// execution that ran none.
|
||||
const response = await request(url)
|
||||
.get(`/api/workflow-executions/${executionId}`)
|
||||
.query({ includeStep: 'true' })
|
||||
.set(authHeader());
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect((response.body as { error: string }).error).toBe('invalid_request');
|
||||
});
|
||||
|
||||
it('returns 400 invalid_request for an includeSteps value that is not a boolean', async () => {
|
||||
const executionId = await createExecution();
|
||||
|
||||
const response = await request(url)
|
||||
.get(`/api/workflow-executions/${executionId}`)
|
||||
.query({ includeSteps: 'yes' })
|
||||
.set(authHeader());
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
|
||||
@@ -24,10 +24,12 @@ export interface ExecutionSnapshot {
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
finishedAt: string | null;
|
||||
/** Only set when the request asked for steps. Oldest first. */
|
||||
steps?: StepDetail[];
|
||||
}
|
||||
|
||||
/**
|
||||
* A step's detail as `GET /:id/steps` reports it.
|
||||
* A step's detail as `GET /:id?includeSteps=true` reports it.
|
||||
*
|
||||
* No `input`: step inputs are re-derived at run time from predecessor outputs
|
||||
* and never persisted, so there's nothing to return. No `attempt`: the engine
|
||||
@@ -43,8 +45,3 @@ export interface StepDetail {
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
/** An envelope, not a bare array, so the response can grow (e.g. pagination). */
|
||||
export interface ExecutionStepsResponse {
|
||||
steps: StepDetail[];
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
export { createEngineServer } from './create-engine-server';
|
||||
export type { EngineServerDeps } from './create-engine-server';
|
||||
export type { ExecutionSnapshot, ExecutionStepsResponse, StepDetail } from './api.types';
|
||||
export type { ExecutionSnapshot, StepDetail } from './api.types';
|
||||
export { fail } from './error-response';
|
||||
export type { EngineErrorResponse } from './error-response';
|
||||
|
||||
@@ -5,13 +5,21 @@ import {
|
||||
ExecutionNotFoundError,
|
||||
type ExecutionQueryService,
|
||||
type ExecutionView,
|
||||
type ExecutionWithStepsView,
|
||||
type StepView,
|
||||
} from '../../execution';
|
||||
import type { ExecutionSnapshot, ExecutionStepsResponse, StepDetail } from '../api.types';
|
||||
import type { ExecutionSnapshot, StepDetail } from '../api.types';
|
||||
import { fail } from '../error-response';
|
||||
|
||||
const ExecutionIdParams = z.object({ id: z.string().uuid() });
|
||||
|
||||
/**
|
||||
* `strict`, because `z.object` strips an unknown key. A misspelled flag would
|
||||
* otherwise answer 200 with no steps, which a caller cannot tell from an
|
||||
* execution that ran none.
|
||||
*/
|
||||
const GetExecutionQuery = z.object({ includeSteps: z.enum(['true', 'false']).optional() }).strict();
|
||||
|
||||
/** The validated `:id`, or `null` once the 400 has been sent. */
|
||||
function parseExecutionId(req: Request, res: Response): string | null {
|
||||
const parsed = ExecutionIdParams.safeParse(req.params);
|
||||
@@ -20,7 +28,7 @@ function parseExecutionId(req: Request, res: Response): string | null {
|
||||
return null;
|
||||
}
|
||||
|
||||
function toExecutionSnapshot(record: ExecutionView): ExecutionSnapshot {
|
||||
function toExecutionSnapshot(record: ExecutionView | ExecutionWithStepsView): ExecutionSnapshot {
|
||||
return {
|
||||
id: record.id,
|
||||
workflowId: record.workflowId,
|
||||
@@ -30,6 +38,9 @@ function toExecutionSnapshot(record: ExecutionView): ExecutionSnapshot {
|
||||
createdAt: record.createdAt.toISOString(),
|
||||
updatedAt: record.updatedAt.toISOString(),
|
||||
finishedAt: record.finishedAt?.toISOString() ?? null,
|
||||
// Absent unless the request asked for steps, which a caller cannot tell
|
||||
// from an execution that ran none.
|
||||
...('steps' in record ? { steps: record.steps.map(toStepDetail) } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -51,9 +62,20 @@ export function createGetExecutionHandler(executionQuery: ExecutionQueryService)
|
||||
const id = parseExecutionId(req, res);
|
||||
if (id === null) return;
|
||||
|
||||
const query = GetExecutionQuery.safeParse(req.query);
|
||||
if (!query.success) {
|
||||
fail(res, 400, { error: 'invalid_request', details: query.error.flatten() });
|
||||
return;
|
||||
}
|
||||
|
||||
// The steps ride along, to save the caller a second round trip, and are read
|
||||
// in the same query so the status cannot predate them.
|
||||
let execution: ExecutionView | ExecutionWithStepsView;
|
||||
try {
|
||||
const execution = await executionQuery.getExecution(id);
|
||||
res.status(200).json(toExecutionSnapshot(execution));
|
||||
execution =
|
||||
query.data.includeSteps === 'true'
|
||||
? await executionQuery.getExecutionWithSteps(id)
|
||||
: await executionQuery.getExecution(id);
|
||||
} catch (error) {
|
||||
if (error instanceof ExecutionNotFoundError) {
|
||||
fail(res, 404, { error: 'not_found' });
|
||||
@@ -61,18 +83,7 @@ export function createGetExecutionHandler(executionQuery: ExecutionQueryService)
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export function createGetExecutionStepsHandler(
|
||||
executionQuery: ExecutionQueryService,
|
||||
): RequestHandler {
|
||||
return async (req, res) => {
|
||||
const id = parseExecutionId(req, res);
|
||||
if (id === null) return;
|
||||
|
||||
const steps = await executionQuery.getSteps(id);
|
||||
const body: ExecutionStepsResponse = { steps: steps.map(toStepDetail) };
|
||||
res.status(200).json(body);
|
||||
res.status(200).json(toExecutionSnapshot(execution));
|
||||
};
|
||||
}
|
||||
|
||||
@@ -2,10 +2,7 @@ import { UUID_V7_PATTERN } from '@n8n/constants';
|
||||
import { Router, type Router as RouterType } from 'express';
|
||||
import { z } from 'zod';
|
||||
|
||||
import {
|
||||
createGetExecutionHandler,
|
||||
createGetExecutionStepsHandler,
|
||||
} from './workflow-executions.handlers';
|
||||
import { createGetExecutionHandler } from './workflow-executions.handlers';
|
||||
import { AdmittanceRejectedError } from '../../admittance';
|
||||
import { jsonValueSchema, UnimplementedError } from '../../common';
|
||||
import { GraphValidationError, MAX_SLOT_INDEX } from '../../graph';
|
||||
@@ -81,7 +78,5 @@ export function createWorkflowExecutionsRouter(deps: EngineServerDeps): RouterTy
|
||||
|
||||
router.get('/:id', createGetExecutionHandler(deps.executionQuery));
|
||||
|
||||
router.get('/:id/steps', createGetExecutionStepsHandler(deps.executionQuery));
|
||||
|
||||
return router;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,335 @@
|
||||
import type { StepDetail, StepStatus, WorkflowGraph } from '@n8n/engine';
|
||||
import type { ExecutionStatus } from 'n8n-workflow';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { toV1RunExecutionData } from '../v1-execution-read';
|
||||
|
||||
// Trigger ──► Set ──► Tail
|
||||
const graph: WorkflowGraph = {
|
||||
nodes: [
|
||||
{ id: 't', name: 'Trigger', type: 'trigger' },
|
||||
{ id: 's', name: 'Set', type: 'v1-node' },
|
||||
{ id: 'x', name: 'Tail', type: 'v1-node' },
|
||||
],
|
||||
edges: [
|
||||
{ from: 't', to: 's', outputIndex: 0, inputIndex: 0 },
|
||||
{ from: 's', to: 'x', outputIndex: 0, inputIndex: 0 },
|
||||
],
|
||||
};
|
||||
|
||||
const step = (overrides: Partial<StepDetail> = {}): StepDetail => ({
|
||||
id: 'step-1',
|
||||
nodeId: 't',
|
||||
iteration: 0,
|
||||
status: 'completed',
|
||||
outputs: [[{ json: { hello: 'world' } }]],
|
||||
error: null,
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
updatedAt: '2026-08-25T10:00:00.250Z',
|
||||
...overrides,
|
||||
});
|
||||
|
||||
describe('toV1RunExecutionData', () => {
|
||||
it('keys run data by node name, with the step outputs under `main`', () => {
|
||||
const data = toV1RunExecutionData(graph, [
|
||||
step(),
|
||||
step({ id: 'step-2', nodeId: 's', outputs: [[{ json: { n: 1 } }, { json: { n: 2 } }]] }),
|
||||
]);
|
||||
|
||||
expect(Object.keys(data.resultData.runData)).toEqual(['Trigger', 'Set']);
|
||||
expect(data.resultData.runData.Trigger[0].data).toEqual({
|
||||
main: [[{ json: { hello: 'world' } }]],
|
||||
});
|
||||
expect(data.resultData.runData.Set[0].data).toEqual({
|
||||
main: [[{ json: { n: 1 } }, { json: { n: 2 } }]],
|
||||
});
|
||||
});
|
||||
|
||||
it('reports no run data for an execution with no steps', () => {
|
||||
const data = toV1RunExecutionData(graph, []);
|
||||
|
||||
expect(data.resultData.runData).toEqual({});
|
||||
expect(data.resultData.error).toBeUndefined();
|
||||
expect(data.resultData.lastNodeExecuted).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each<[StepStatus, ExecutionStatus]>([
|
||||
['completed', 'success'],
|
||||
['failed', 'error'],
|
||||
['running', 'running'],
|
||||
])('maps step status %j to %j', (status, expected) => {
|
||||
const data = toV1RunExecutionData(graph, [step({ status })]);
|
||||
|
||||
expect(data.resultData.runData.Trigger[0].executionStatus).toBe(expected);
|
||||
});
|
||||
|
||||
// `cancelQueuedSteps` is the only writer of `cancelled`, and it updates queued
|
||||
// rows only, so a cancelled step never ran.
|
||||
it.each<StepStatus>(['queued', 'skipped', 'cancelled'])(
|
||||
'reports no run for a %j step, the way v1 reports a node that did not run',
|
||||
(status) => {
|
||||
const data = toV1RunExecutionData(graph, [step({ status })]);
|
||||
|
||||
expect(data.resultData.runData).toEqual({});
|
||||
},
|
||||
);
|
||||
|
||||
it('takes timing from the row timestamps', () => {
|
||||
const data = toV1RunExecutionData(graph, [step()]);
|
||||
|
||||
expect(data.resultData.runData.Trigger[0]).toMatchObject({
|
||||
startTime: Date.parse('2026-08-25T10:00:00.000Z'),
|
||||
executionTime: 250,
|
||||
});
|
||||
});
|
||||
|
||||
it('clamps a negative duration to zero', () => {
|
||||
const data = toV1RunExecutionData(graph, [
|
||||
step({ createdAt: '2026-08-25T10:00:01.000Z', updatedAt: '2026-08-25T10:00:00.000Z' }),
|
||||
]);
|
||||
|
||||
expect(data.resultData.runData.Trigger[0].executionTime).toBe(0);
|
||||
});
|
||||
|
||||
it('numbers `executionIndex` by data plane order, not by iteration', () => {
|
||||
const data = toV1RunExecutionData(graph, [
|
||||
step(),
|
||||
step({ id: 'step-2', nodeId: 's' }),
|
||||
step({ id: 'step-3', nodeId: 'x' }),
|
||||
]);
|
||||
|
||||
expect(data.resultData.runData.Trigger[0].executionIndex).toBe(0);
|
||||
expect(data.resultData.runData.Set[0].executionIndex).toBe(1);
|
||||
expect(data.resultData.runData.Tail[0].executionIndex).toBe(2);
|
||||
});
|
||||
|
||||
it('rebuilds `source` from the graph edges, and leaves it empty for a trigger', () => {
|
||||
const data = toV1RunExecutionData(graph, [step(), step({ id: 'step-2', nodeId: 's' })]);
|
||||
|
||||
expect(data.resultData.runData.Trigger[0].source).toEqual([]);
|
||||
expect(data.resultData.runData.Set[0].source).toEqual([
|
||||
{ previousNode: 'Trigger', previousNodeOutput: 0 },
|
||||
]);
|
||||
});
|
||||
|
||||
it('pads an unfilled input slot with null', () => {
|
||||
const merge: WorkflowGraph = {
|
||||
nodes: [
|
||||
{ id: 't', name: 'Trigger', type: 'trigger' },
|
||||
{ id: 'm', name: 'Merge', type: 'v1-node' },
|
||||
],
|
||||
// Only the second input is connected.
|
||||
edges: [{ from: 't', to: 'm', outputIndex: 0, inputIndex: 1 }],
|
||||
};
|
||||
|
||||
const data = toV1RunExecutionData(merge, [step({ nodeId: 'm' })]);
|
||||
|
||||
expect(data.resultData.runData.Merge[0].source).toEqual([
|
||||
null,
|
||||
{ previousNode: 'Trigger', previousNodeOutput: 0 },
|
||||
]);
|
||||
});
|
||||
|
||||
it('drops a step whose node is no longer in the graph', () => {
|
||||
const data = toV1RunExecutionData(graph, [step({ nodeId: 'gone' })]);
|
||||
|
||||
expect(data.resultData.runData).toEqual({});
|
||||
});
|
||||
|
||||
it('reports a failed step as the execution error and the last node executed', () => {
|
||||
const data = toV1RunExecutionData(graph, [
|
||||
step(),
|
||||
step({
|
||||
id: 'step-2',
|
||||
nodeId: 's',
|
||||
status: 'failed',
|
||||
outputs: null,
|
||||
error: { name: 'NodeOperationError', message: 'Boom', stack: 'at somewhere' },
|
||||
}),
|
||||
]);
|
||||
|
||||
const task = data.resultData.runData.Set[0];
|
||||
expect(task.data).toBeUndefined();
|
||||
expect(task.error).toMatchObject({ name: 'NodeOperationError', message: 'Boom' });
|
||||
expect(data.resultData.error).toBe(task.error);
|
||||
expect(data.resultData.lastNodeExecuted).toBe('Set');
|
||||
});
|
||||
|
||||
it('reports the last step as the last node executed when nothing failed', () => {
|
||||
const data = toV1RunExecutionData(graph, [step(), step({ id: 'step-2', nodeId: 's' })]);
|
||||
|
||||
expect(data.resultData.error).toBeUndefined();
|
||||
expect(data.resultData.lastNodeExecuted).toBe('Set');
|
||||
});
|
||||
|
||||
it('fills a gap in the iteration sequence rather than leaving a hole', () => {
|
||||
const data = toV1RunExecutionData(graph, [
|
||||
step({ nodeId: 's', iteration: 0 }),
|
||||
step({ id: 'step-3', nodeId: 's', iteration: 2 }),
|
||||
]);
|
||||
|
||||
const runs = data.resultData.runData.Set;
|
||||
expect(runs).toHaveLength(3);
|
||||
expect(runs[1]).toEqual({
|
||||
startTime: 0,
|
||||
executionTime: 0,
|
||||
executionIndex: 1,
|
||||
source: [],
|
||||
data: { main: [[]] },
|
||||
});
|
||||
expect(runs[2].data).toEqual({ main: [[{ json: { hello: 'world' } }]] });
|
||||
});
|
||||
|
||||
it('carries no resume token, since a v2 execution has none', () => {
|
||||
expect(toV1RunExecutionData(graph, [step()]).resumeToken).toBe('');
|
||||
});
|
||||
|
||||
it('carries no execution data, since a read reports no runtime state', () => {
|
||||
expect(toV1RunExecutionData(graph, [step()]).executionData).toBeUndefined();
|
||||
});
|
||||
|
||||
describe('parallel branches', () => {
|
||||
// Two branches off the trigger. Row creation order says nothing about
|
||||
// which branch settled first.
|
||||
const forked: WorkflowGraph = {
|
||||
nodes: [
|
||||
{ id: 't', name: 'Trigger', type: 'trigger' },
|
||||
{ id: 'a', name: 'Slow', type: 'v1-node' },
|
||||
{ id: 'b', name: 'Fast', type: 'v1-node' },
|
||||
],
|
||||
edges: [
|
||||
{ from: 't', to: 'a', outputIndex: 0, inputIndex: 0 },
|
||||
{ from: 't', to: 'b', outputIndex: 1, inputIndex: 0 },
|
||||
],
|
||||
};
|
||||
|
||||
it('reports the step that settled last as the last node executed', () => {
|
||||
const data = toV1RunExecutionData(forked, [
|
||||
// Created first, settled last.
|
||||
step({ nodeId: 'a', updatedAt: '2026-08-25T10:00:09.000Z' }),
|
||||
step({ id: 'step-2', nodeId: 'b', updatedAt: '2026-08-25T10:00:01.000Z' }),
|
||||
]);
|
||||
|
||||
expect(data.resultData.lastNodeExecuted).toBe('Slow');
|
||||
});
|
||||
|
||||
it('reports the failure that settled first as the execution error', () => {
|
||||
const data = toV1RunExecutionData(forked, [
|
||||
step({
|
||||
nodeId: 'a',
|
||||
status: 'failed',
|
||||
outputs: null,
|
||||
error: { name: 'NodeOperationError', message: 'Second' },
|
||||
updatedAt: '2026-08-25T10:00:09.000Z',
|
||||
}),
|
||||
step({
|
||||
id: 'step-2',
|
||||
nodeId: 'b',
|
||||
status: 'failed',
|
||||
outputs: null,
|
||||
error: { name: 'NodeOperationError', message: 'First' },
|
||||
updatedAt: '2026-08-25T10:00:01.000Z',
|
||||
}),
|
||||
]);
|
||||
|
||||
// The first failure is the one that stopped the run.
|
||||
expect(data.resultData.error).toMatchObject({ message: 'First' });
|
||||
expect(data.resultData.lastNodeExecuted).toBe('Fast');
|
||||
});
|
||||
});
|
||||
|
||||
describe('loop lineage', () => {
|
||||
// Trigger ──► Loop ──o1──► Body ──(back)──► Loop
|
||||
// └──o0──► Done
|
||||
const looped: WorkflowGraph = {
|
||||
nodes: [
|
||||
{ id: 't', name: 'Trigger', type: 'trigger' },
|
||||
{ id: 'loop', name: 'Loop', type: 'batch' },
|
||||
{ id: 'body', name: 'Body', type: 'v1-node' },
|
||||
{ id: 'done', name: 'Done', type: 'v1-node' },
|
||||
],
|
||||
edges: [
|
||||
{ from: 't', to: 'loop', outputIndex: 0, inputIndex: 0 },
|
||||
{ from: 'loop', to: 'body', outputIndex: 1, inputIndex: 0 },
|
||||
{ from: 'body', to: 'loop', outputIndex: 0, inputIndex: 0, isBackEdge: true },
|
||||
{ from: 'loop', to: 'done', outputIndex: 0, inputIndex: 0 },
|
||||
],
|
||||
};
|
||||
|
||||
const twoPasses = [
|
||||
step({ nodeId: 'loop', iteration: 0 }),
|
||||
step({ id: 's2', nodeId: 'body', iteration: 0 }),
|
||||
step({ id: 's3', nodeId: 'loop', iteration: 1 }),
|
||||
step({ id: 's4', nodeId: 'body', iteration: 1 }),
|
||||
step({ id: 's5', nodeId: 'done', iteration: 0 }),
|
||||
];
|
||||
|
||||
it('points a loop member at the same pass of its predecessor', () => {
|
||||
const data = toV1RunExecutionData(looped, twoPasses);
|
||||
|
||||
expect(data.resultData.runData.Body[0].source).toEqual([
|
||||
{ previousNode: 'Loop', previousNodeOutput: 1 },
|
||||
]);
|
||||
expect(data.resultData.runData.Body[1].source).toEqual([
|
||||
{ previousNode: 'Loop', previousNodeOutput: 1, previousNodeRun: 1 },
|
||||
]);
|
||||
});
|
||||
|
||||
it("points the node after a loop at the loop's last pass", () => {
|
||||
const data = toV1RunExecutionData(looped, twoPasses);
|
||||
|
||||
expect(data.resultData.runData.Done[0].source).toEqual([
|
||||
{ previousNode: 'Loop', previousNodeOutput: 0, previousNodeRun: 1 },
|
||||
]);
|
||||
});
|
||||
|
||||
// Trigger ──► Loop ──o1──► Body ──(back)──► Loop
|
||||
// │ └───o1──► Aside
|
||||
// └────────────────────────────────► Aside
|
||||
//
|
||||
// `Aside` still runs on a pass where `Body` is skipped, because its other
|
||||
// input is live.
|
||||
const skippedLastPass: WorkflowGraph = {
|
||||
nodes: [
|
||||
...looped.nodes.filter((node) => node.id !== 'done'),
|
||||
{ id: 'aside', name: 'Aside', type: 'v1-node' },
|
||||
],
|
||||
edges: [
|
||||
...looped.edges.filter((edge) => edge.to !== 'done'),
|
||||
{ from: 'body', to: 'aside', outputIndex: 1, inputIndex: 0 },
|
||||
{ from: 't', to: 'aside', outputIndex: 0, inputIndex: 1 },
|
||||
],
|
||||
};
|
||||
|
||||
it('points a node after a loop at the last pass that was reported', () => {
|
||||
const data = toV1RunExecutionData(skippedLastPass, [
|
||||
step({ nodeId: 'loop', iteration: 0 }),
|
||||
step({ id: 's2', nodeId: 'body', iteration: 0 }),
|
||||
step({ id: 's3', nodeId: 'loop', iteration: 1 }),
|
||||
step({ id: 's4', nodeId: 'body', iteration: 1 }),
|
||||
step({ id: 's5', nodeId: 'loop', iteration: 2 }),
|
||||
// The last pass of the body ran nothing, so it gets no run data.
|
||||
step({ id: 's6', nodeId: 'body', iteration: 2, status: 'skipped', outputs: null }),
|
||||
step({ id: 's7', nodeId: 'aside', iteration: 0 }),
|
||||
]);
|
||||
|
||||
// Pass 2 has no run, so naming it would name a run v1 cannot read.
|
||||
expect(data.resultData.runData.Body).toHaveLength(2);
|
||||
expect(data.resultData.runData.Aside[0].source[0]).toEqual({
|
||||
previousNode: 'Body',
|
||||
previousNodeOutput: 1,
|
||||
previousNodeRun: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it('points the loop entry at run 0, since the predecessor ran once', () => {
|
||||
const data = toV1RunExecutionData(looped, twoPasses);
|
||||
|
||||
// Both passes read the entry edge: a back edge names a source v1 cannot
|
||||
// express, so it is skipped.
|
||||
for (const run of data.resultData.runData.Loop) {
|
||||
expect(run.source).toEqual([{ previousNode: 'Trigger', previousNodeOutput: 0 }]);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -2,6 +2,7 @@ export { V1WorkflowConverter } from './v1-workflow-converter';
|
||||
export { V1StepExecutor } from './v1-step-executor';
|
||||
export { createEngineStepDataLoader } from './engine-step-data-loader';
|
||||
export { fromStepInputs, toStepOutputs } from './io';
|
||||
export { toV1RunExecutionData } from './v1-execution-read';
|
||||
export {
|
||||
AmbiguousTriggerError,
|
||||
NotATriggerError,
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
import { isTriggerStepConfig, isV1NodeStepConfig } from './guards';
|
||||
import { fromStepInputs } from './io';
|
||||
import type { CreateExecuteContextParams, V1Execution, V1NodeStepConfig } from './types';
|
||||
import { emptyRun, forwardEdgesByTarget, nodeNamesById, toSourceSlots } from './v1-run-data';
|
||||
|
||||
export function toV1Execution(
|
||||
graph: WorkflowGraph,
|
||||
@@ -59,7 +60,7 @@ function toV1Nodes(graph: WorkflowGraph): INode[] {
|
||||
}
|
||||
|
||||
function toV1Connections(graph: WorkflowGraph): IConnections {
|
||||
const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
|
||||
const namesById = nodeNamesById(graph);
|
||||
|
||||
const connections: IConnections = {};
|
||||
for (const edge of graph.edges) {
|
||||
@@ -88,7 +89,7 @@ function toV1RunData(
|
||||
activeNodeId: string,
|
||||
activeIteration: number,
|
||||
): IRunData {
|
||||
const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
|
||||
const namesById = nodeNamesById(graph);
|
||||
const sourcesByNodeId = toV1Sources(graph);
|
||||
const activeLoop = deriveLoops(graph).find((loop) => loop.memberIds.has(activeNodeId));
|
||||
|
||||
@@ -111,37 +112,31 @@ function toV1RunData(
|
||||
|
||||
const source = sourcesByNodeId.get(completedNodeId) ?? [];
|
||||
|
||||
// A pass the node was skipped on gets an empty run rather than a gap. v1
|
||||
// reads whatever entry it finds, and a gap would crash it. The empty run
|
||||
// holds one empty slot, since zero slots reads to v1 as no data at all.
|
||||
runData[nodeName] = Array.from({ length: lastShown + 1 }, (_, iteration) => ({
|
||||
startTime: 0,
|
||||
executionTime: 0,
|
||||
executionIndex: iteration,
|
||||
source,
|
||||
data: {
|
||||
[MAIN_CONNECTION_TYPE]: fromStepInputs(outputsByIteration[iteration] ?? [[]]),
|
||||
},
|
||||
}));
|
||||
// A pass the node was skipped on gets an empty run rather than a gap.
|
||||
runData[nodeName] = Array.from({ length: lastShown + 1 }, (_, iteration) => {
|
||||
const outputs = outputsByIteration[iteration];
|
||||
if (outputs === undefined) return emptyRun(iteration, source);
|
||||
|
||||
return {
|
||||
startTime: 0,
|
||||
executionTime: 0,
|
||||
executionIndex: iteration,
|
||||
source,
|
||||
data: { [MAIN_CONNECTION_TYPE]: fromStepInputs(outputs) },
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
return runData;
|
||||
}
|
||||
|
||||
/** The execute path reports one pass, so no source names a `previousNodeRun`. */
|
||||
export function toV1Sources(graph: WorkflowGraph): Map<string, Array<ISourceData | null>> {
|
||||
const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
|
||||
const namesById = nodeNamesById(graph);
|
||||
|
||||
const sourcesByNodeId = new Map<string, Array<ISourceData | null>>();
|
||||
for (const edge of graph.edges) {
|
||||
if (edge.isBackEdge === true) continue;
|
||||
const fromName = namesById.get(edge.from);
|
||||
if (fromName === undefined) continue;
|
||||
|
||||
const source = sourcesByNodeId.get(edge.to) ?? [];
|
||||
const inputIndex = edge.inputIndex ?? 0;
|
||||
while (source.length <= inputIndex) source.push(null);
|
||||
source[inputIndex] ??= { previousNode: fromName, previousNodeOutput: edge.outputIndex };
|
||||
sourcesByNodeId.set(edge.to, source);
|
||||
for (const [nodeId, edges] of forwardEdgesByTarget(graph)) {
|
||||
sourcesByNodeId.set(nodeId, toSourceSlots(edges, namesById));
|
||||
}
|
||||
return sourcesByNodeId;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
/**
|
||||
* Turns an engine v2 execution back into the v1 `IRunExecutionData` the editor
|
||||
* renders. Read path only. `toV1RunData` in `./v1-adapters` serves one step's
|
||||
* expression context and trims loops to the active pass; this reports every
|
||||
* iteration.
|
||||
*/
|
||||
|
||||
import { deriveLoops } from '@n8n/engine';
|
||||
import type {
|
||||
GraphEdge,
|
||||
StepDetail,
|
||||
StepError,
|
||||
StepStatus,
|
||||
WorkflowGraph,
|
||||
WorkflowLoop,
|
||||
} from '@n8n/engine';
|
||||
import type {
|
||||
ExecutionError,
|
||||
ExecutionStatus,
|
||||
IRunData,
|
||||
ISourceData,
|
||||
ITaskData,
|
||||
} from 'n8n-workflow';
|
||||
import { createRunExecutionData, WorkflowOperationError } from 'n8n-workflow';
|
||||
|
||||
import { MAIN_CONNECTION_TYPE } from './constants';
|
||||
import { fromStepInputs } from './io';
|
||||
import { emptyRun, forwardEdgesByTarget, nodeNamesById, toSourceSlots } from './v1-run-data';
|
||||
|
||||
/**
|
||||
* Only a step that ran gets an entry. v1 reports a node that did not run by
|
||||
* having no `runData` for it, and an entry would make the canvas claim it ran.
|
||||
* `queued` never started, `skipped` was decided against, and `cancelled` is
|
||||
* reachable only through `cancelQueuedSteps`, which cancels queued rows.
|
||||
*/
|
||||
const TASK_STATUS_V1 = new Map<StepStatus, ExecutionStatus>([
|
||||
['completed', 'success'],
|
||||
['failed', 'error'],
|
||||
['running', 'running'],
|
||||
]);
|
||||
|
||||
interface StepRun {
|
||||
step: StepDetail;
|
||||
nodeName: string;
|
||||
taskData: ITaskData;
|
||||
}
|
||||
|
||||
/** What a step's input lineage needs, prepared once for the whole execution. */
|
||||
interface Lineage {
|
||||
namesById: Map<string, string>;
|
||||
/** Forward edges only, by target node. */
|
||||
edgesByTarget: Map<string, GraphEdge[]>;
|
||||
loops: WorkflowLoop[];
|
||||
lastIterationByNodeId: Map<string, number>;
|
||||
}
|
||||
|
||||
export function toV1RunExecutionData(graph: WorkflowGraph, steps: StepDetail[]) {
|
||||
const runs = toStepRuns(graph, steps);
|
||||
|
||||
// By settle time, not by creation order: parallel branches settle
|
||||
// independently, so the row created last is not the step that ran last.
|
||||
const bySettleTime = [...runs].sort((a, b) => settledAt(a) - settledAt(b));
|
||||
// The failure that ended the execution is the first one to settle.
|
||||
const failed = bySettleTime.find((run) => run.step.status === 'failed');
|
||||
|
||||
return createRunExecutionData({
|
||||
resultData: {
|
||||
runData: toRunData(runs),
|
||||
error: failed?.taskData.error,
|
||||
lastNodeExecuted: (failed ?? bySettleTime.at(-1))?.nodeName,
|
||||
},
|
||||
// A read reports what ran. The factory otherwise fills in the runtime
|
||||
// structures a run needs, such as the node execution stack.
|
||||
executionData: null,
|
||||
// A v2 execution has no resume token. The factory otherwise mints one.
|
||||
resumeToken: '',
|
||||
});
|
||||
}
|
||||
|
||||
/** When the step reached its current status. `createdAt` is when it was planned. */
|
||||
function settledAt(run: StepRun): number {
|
||||
return Date.parse(run.step.updatedAt);
|
||||
}
|
||||
|
||||
/** @param steps Oldest first, as the data plane orders them. Position is the run order. */
|
||||
function toStepRuns(graph: WorkflowGraph, steps: StepDetail[]): StepRun[] {
|
||||
const lineage = toLineage(graph, steps);
|
||||
|
||||
const runs: StepRun[] = [];
|
||||
for (const step of steps) {
|
||||
const executionStatus = TASK_STATUS_V1.get(step.status);
|
||||
if (executionStatus === undefined) continue;
|
||||
|
||||
const nodeName = lineage.namesById.get(step.nodeId);
|
||||
// v1 keys run data by node name only.
|
||||
if (nodeName === undefined) continue;
|
||||
|
||||
runs.push({
|
||||
step,
|
||||
nodeName,
|
||||
taskData: toTaskData(step, executionStatus, runs.length, toSources(lineage, step)),
|
||||
});
|
||||
}
|
||||
|
||||
return runs;
|
||||
}
|
||||
|
||||
function toLineage(graph: WorkflowGraph, steps: StepDetail[]): Lineage {
|
||||
const lastIterationByNodeId = new Map<string, number>();
|
||||
for (const step of steps) {
|
||||
// Only a reported pass counts. A pass v1 never hears about, such as a body
|
||||
// step skipped on the last pass, would name a run that has no `runData`.
|
||||
if (!TASK_STATUS_V1.has(step.status)) continue;
|
||||
const seen = lastIterationByNodeId.get(step.nodeId);
|
||||
if (seen === undefined || step.iteration > seen) {
|
||||
lastIterationByNodeId.set(step.nodeId, step.iteration);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
namesById: nodeNamesById(graph),
|
||||
edgesByTarget: forwardEdgesByTarget(graph),
|
||||
loops: deriveLoops(graph),
|
||||
lastIterationByNodeId,
|
||||
};
|
||||
}
|
||||
|
||||
/** Unlike the execute path, a read reports which pass of the predecessor fed each one. */
|
||||
function toSources(lineage: Lineage, step: StepDetail): Array<ISourceData | null> {
|
||||
return toSourceSlots(lineage.edgesByTarget.get(step.nodeId) ?? [], lineage.namesById, (edge) =>
|
||||
toPreviousNodeRun(lineage, edge, step.iteration),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Which run of the predecessor fed this pass. The cases follow `classifyEdge`
|
||||
* in the engine's `iteration-mapping`: an edge inside one loop keeps the pass,
|
||||
* an edge leaving a loop reads its last pass, and anything else reads run 0.
|
||||
*/
|
||||
function toPreviousNodeRun(lineage: Lineage, edge: GraphEdge, iteration: number): number {
|
||||
const sourceLoop = lineage.loops.find((loop) => loop.memberIds.has(edge.from));
|
||||
// `plain` or `entry`: the predecessor ran once.
|
||||
if (!sourceLoop) return 0;
|
||||
// `intra`: both ends share a pass.
|
||||
if (sourceLoop.memberIds.has(edge.to)) return iteration;
|
||||
// `exit`: the predecessor's terminal pass.
|
||||
return lineage.lastIterationByNodeId.get(edge.from) ?? 0;
|
||||
}
|
||||
|
||||
function toTaskData(
|
||||
step: StepDetail,
|
||||
executionStatus: ExecutionStatus,
|
||||
executionIndex: number,
|
||||
source: Array<ISourceData | null>,
|
||||
): ITaskData {
|
||||
// TODO(CAT-4234): report real run timing. Row timestamps include queue time.
|
||||
const startTime = Date.parse(step.createdAt);
|
||||
|
||||
return {
|
||||
startTime,
|
||||
executionTime: Math.max(0, Date.parse(step.updatedAt) - startTime),
|
||||
executionIndex,
|
||||
source,
|
||||
executionStatus,
|
||||
// Every output slot is `main`: no other connection type exists here.
|
||||
...(step.outputs === null
|
||||
? {}
|
||||
: { data: { [MAIN_CONNECTION_TYPE]: fromStepInputs(step.outputs) } }),
|
||||
...(step.error === null ? {} : { error: toV1Error(step.error) }),
|
||||
};
|
||||
}
|
||||
|
||||
function toRunData(runs: StepRun[]): IRunData {
|
||||
const runData: IRunData = {};
|
||||
|
||||
for (const { step, nodeName, taskData } of runs) {
|
||||
const tasks = (runData[nodeName] ??= []);
|
||||
while (tasks.length < step.iteration) {
|
||||
tasks.push(emptyRun(tasks.length));
|
||||
}
|
||||
tasks[step.iteration] = taskData;
|
||||
}
|
||||
|
||||
return runData;
|
||||
}
|
||||
|
||||
/**
|
||||
* The data plane holds nothing v1-shaped, so this rebuilds the minimum. Only
|
||||
* `name` is worth restoring: `ExecutionBaseError.toJSON` drops `stack`.
|
||||
*/
|
||||
function toV1Error(error: StepError): ExecutionError {
|
||||
const v1Error = new WorkflowOperationError(error.message);
|
||||
v1Error.name = error.name;
|
||||
return v1Error;
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
/**
|
||||
* The pieces both v1 run-data builders share: node names, forward edges, input
|
||||
* sources and the empty run. `v1-adapters` builds run data for one step's
|
||||
* expression context; `v1-execution-read` builds it for the whole execution.
|
||||
* They differ in what they report per pass, not in how a graph maps to v1.
|
||||
*/
|
||||
|
||||
import type { GraphEdge, WorkflowGraph } from '@n8n/engine';
|
||||
import type { ISourceData, ITaskData } from 'n8n-workflow';
|
||||
|
||||
import { MAIN_CONNECTION_TYPE } from './constants';
|
||||
|
||||
export function nodeNamesById(graph: WorkflowGraph): Map<string, string> {
|
||||
return new Map(graph.nodes.map((node) => [node.id, node.name]));
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward edges by target node. A back edge's source is the previous pass, a
|
||||
* lineage v1 cannot express, so it reads the forward edge into the loop entry
|
||||
* instead.
|
||||
*/
|
||||
export function forwardEdgesByTarget(graph: WorkflowGraph): Map<string, GraphEdge[]> {
|
||||
const edgesByTarget = new Map<string, GraphEdge[]>();
|
||||
|
||||
for (const edge of graph.edges) {
|
||||
if (edge.isBackEdge === true) continue;
|
||||
const edges = edgesByTarget.get(edge.to);
|
||||
if (edges) edges.push(edge);
|
||||
else edgesByTarget.set(edge.to, [edge]);
|
||||
}
|
||||
|
||||
return edgesByTarget;
|
||||
}
|
||||
|
||||
/**
|
||||
* One entry for each input slot of one node, indexed by slot. Empty for a node
|
||||
* with no input, `null` for an unconnected slot.
|
||||
*
|
||||
* @param edges The forward edges into the node.
|
||||
* @param previousNodeRun Which run of the predecessor fed this one. Left out
|
||||
* when the caller reports a single pass, since v1 reads a missing value as 0.
|
||||
*/
|
||||
export function toSourceSlots(
|
||||
edges: GraphEdge[],
|
||||
namesById: Map<string, string>,
|
||||
previousNodeRun?: (edge: GraphEdge) => number,
|
||||
): Array<ISourceData | null> {
|
||||
const sources: Array<ISourceData | null> = [];
|
||||
|
||||
for (const edge of edges) {
|
||||
const previousNode = namesById.get(edge.from);
|
||||
if (previousNode === undefined) continue;
|
||||
|
||||
const inputIndex = edge.inputIndex ?? 0;
|
||||
while (sources.length <= inputIndex) sources.push(null);
|
||||
// First edge into a slot wins.
|
||||
if (sources[inputIndex] !== null) continue;
|
||||
|
||||
const run = previousNodeRun?.(edge) ?? 0;
|
||||
sources[inputIndex] = {
|
||||
previousNode,
|
||||
previousNodeOutput: edge.outputIndex,
|
||||
// v1 reads a missing value as 0, so only a real loop pass is reported.
|
||||
...(run === 0 ? {} : { previousNodeRun: run }),
|
||||
};
|
||||
}
|
||||
|
||||
return sources;
|
||||
}
|
||||
|
||||
/**
|
||||
* A pass the node recorded nothing for. v1 reads whatever `ITaskData[]` entry it
|
||||
* finds, so a gap would crash it. The run holds one empty slot, since zero slots
|
||||
* reads to v1 as no data at all.
|
||||
*/
|
||||
export function emptyRun(
|
||||
executionIndex: number,
|
||||
source: Array<ISourceData | null> = [],
|
||||
): ITaskData {
|
||||
return {
|
||||
startTime: 0,
|
||||
executionTime: 0,
|
||||
executionIndex,
|
||||
source,
|
||||
data: { [MAIN_CONNECTION_TYPE]: [[]] },
|
||||
};
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { WorkflowEntity, WorkflowRepository } from '@n8n/db';
|
||||
import type { ExecutionSnapshot, ExecutionStatus } from '@n8n/engine';
|
||||
import type { ExecutionSnapshot, ExecutionStatus, StepDetail } from '@n8n/engine';
|
||||
import type { ExecutionStatus as ExecutionStatusV1 } from 'n8n-workflow';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
@@ -11,12 +11,24 @@ import type { ExecutionIdV2 } from '../execution-id';
|
||||
const EXECUTION_ID = '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa' as ExecutionIdV2;
|
||||
const WORKFLOW_ID = 'wf-1';
|
||||
|
||||
const step = (overrides: Partial<StepDetail> = {}): StepDetail => ({
|
||||
id: 'step-1',
|
||||
nodeId: 'trigger-id',
|
||||
iteration: 0,
|
||||
status: 'completed',
|
||||
outputs: [[{ json: { hello: 'world' } }]],
|
||||
error: null,
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
updatedAt: '2026-08-25T10:00:00.000Z',
|
||||
...overrides,
|
||||
});
|
||||
|
||||
const snapshot = (overrides: Partial<ExecutionSnapshot> = {}): ExecutionSnapshot => ({
|
||||
id: EXECUTION_ID,
|
||||
workflowId: WORKFLOW_ID,
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
graph: { nodes: [], edges: [] },
|
||||
graph: { nodes: [{ id: 'trigger-id', name: 'Trigger', type: 'trigger' }], edges: [] },
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
updatedAt: '2026-08-25T10:00:05.000Z',
|
||||
finishedAt: '2026-08-25T10:00:05.000Z',
|
||||
@@ -30,7 +42,10 @@ describe('EngineV2ExecutionReader', () => {
|
||||
|
||||
const workflow = mock<WorkflowEntity>({
|
||||
id: WORKFLOW_ID,
|
||||
name: 'v2',
|
||||
nodes: [],
|
||||
connections: {},
|
||||
nodeGroups: [],
|
||||
settings: { engineType: 'v2' },
|
||||
});
|
||||
|
||||
@@ -44,7 +59,7 @@ describe('EngineV2ExecutionReader', () => {
|
||||
it('should map the snapshot onto an execution response', async () => {
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
|
||||
expect(dataPlane.getExecution).toHaveBeenCalledWith(EXECUTION_ID);
|
||||
expect(dataPlane.getExecution).toHaveBeenCalledWith(EXECUTION_ID, { includeSteps: true });
|
||||
expect(result).toEqual({
|
||||
id: EXECUTION_ID,
|
||||
workflowId: WORKFLOW_ID,
|
||||
@@ -55,8 +70,14 @@ describe('EngineV2ExecutionReader', () => {
|
||||
startedAt: new Date('2026-08-25T10:00:00.000Z'),
|
||||
stoppedAt: new Date('2026-08-25T10:00:05.000Z'),
|
||||
storedAt: 'db',
|
||||
data: { version: 1, resultData: { runData: {} } },
|
||||
workflowData: workflow,
|
||||
// The run-data mapping has its own tests. This one checks the envelope.
|
||||
data: expect.objectContaining({
|
||||
version: 1,
|
||||
resultData: expect.objectContaining({ runData: {} }),
|
||||
}),
|
||||
// `mock<T>` adds proxy symbols that break deep equality. The exact
|
||||
// field set has its own test.
|
||||
workflowData: expect.objectContaining({ id: WORKFLOW_ID, name: 'v2' }),
|
||||
customData: {},
|
||||
annotation: { tags: [] },
|
||||
});
|
||||
@@ -72,6 +93,20 @@ describe('EngineV2ExecutionReader', () => {
|
||||
expect(result?.workflowData.settings?.redactionPolicy).toBe('all');
|
||||
});
|
||||
|
||||
it('should report the same narrow workflow projection the v1 path reports', async () => {
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
|
||||
// Anything wider leaks the raw row, such as `shared` and its project.
|
||||
expect(Object.keys(result?.workflowData ?? {}).sort()).toEqual([
|
||||
'connections',
|
||||
'id',
|
||||
'name',
|
||||
'nodeGroups',
|
||||
'nodes',
|
||||
'settings',
|
||||
]);
|
||||
});
|
||||
|
||||
it.each<[ExecutionStatus, ExecutionStatusV1, boolean]>([
|
||||
['queued', 'new', false],
|
||||
['running', 'running', false],
|
||||
@@ -103,6 +138,27 @@ describe('EngineV2ExecutionReader', () => {
|
||||
expect(result?.stoppedAt).toBeUndefined();
|
||||
});
|
||||
|
||||
it('should map the steps onto run data keyed by node name', async () => {
|
||||
dataPlane.getExecution.mockResolvedValue(snapshot({ steps: [step()] }));
|
||||
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
|
||||
expect(result?.data.resultData.runData).toEqual({
|
||||
Trigger: [
|
||||
expect.objectContaining({
|
||||
executionStatus: 'success',
|
||||
data: { main: [[{ json: { hello: 'world' } }]] },
|
||||
}),
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it('should report empty run data when the snapshot carries no steps', async () => {
|
||||
const result = await reader.findOne(EXECUTION_ID, [WORKFLOW_ID]);
|
||||
|
||||
expect(result?.data.resultData.runData).toEqual({});
|
||||
});
|
||||
|
||||
it('should return undefined when the data plane holds no such execution', async () => {
|
||||
dataPlane.getExecution.mockResolvedValue(undefined);
|
||||
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
import type { IExecutionResponse } from '@n8n/db';
|
||||
import type { IExecutionResponse, WorkflowEntity } from '@n8n/db';
|
||||
import { WorkflowRepository } from '@n8n/db';
|
||||
import { Service } from '@n8n/di';
|
||||
import type { ExecutionMode, ExecutionSnapshot, ExecutionStatus } from '@n8n/engine';
|
||||
import type { ExecutionStatus as ExecutionStatusV1, WorkflowExecuteMode } from 'n8n-workflow';
|
||||
import { createEmptyRunExecutionData } from 'n8n-workflow';
|
||||
import type {
|
||||
ExecutionStatus as ExecutionStatusV1,
|
||||
IRunExecutionData,
|
||||
WorkflowExecuteMode,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
|
||||
|
||||
import { toWorkflowSnapshot } from './execution-data/types';
|
||||
import type { ExecutionIdV2 } from './execution-id';
|
||||
|
||||
/** A status added later reads as `unknown` rather than being guessed at. */
|
||||
@@ -42,7 +46,7 @@ export class EngineV2ExecutionReader {
|
||||
): Promise<IExecutionResponse | undefined> {
|
||||
// TODO(CAT-4235): mirror this metadata on the control plane, so we can
|
||||
// authorize before reading.
|
||||
const snapshot = await this.dataPlane.getExecution(executionId);
|
||||
const snapshot = await this.dataPlane.getExecution(executionId, { includeSteps: true });
|
||||
if (!snapshot) return undefined;
|
||||
|
||||
// The `workflow:read` check.
|
||||
@@ -51,12 +55,21 @@ export class EngineV2ExecutionReader {
|
||||
const workflowData = await this.workflowRepository.findById(snapshot.workflowId);
|
||||
if (!workflowData) return undefined;
|
||||
|
||||
return this.toExecutionResponse(snapshot, workflowData);
|
||||
// Lazily imported: a top-level import would pull `@n8n/engine` into every
|
||||
// n8n process, including ones with the module off.
|
||||
const { toV1RunExecutionData } = await import('@n8n/node-engine-compatibility');
|
||||
|
||||
return this.toExecutionResponse(
|
||||
snapshot,
|
||||
workflowData,
|
||||
toV1RunExecutionData(snapshot.graph, snapshot.steps ?? []),
|
||||
);
|
||||
}
|
||||
|
||||
private toExecutionResponse(
|
||||
snapshot: ExecutionSnapshot,
|
||||
workflowData: IExecutionResponse['workflowData'],
|
||||
workflow: WorkflowEntity,
|
||||
data: IRunExecutionData,
|
||||
): IExecutionResponse {
|
||||
// No real run timing yet (CAT-4234), so both come from the row.
|
||||
const startedAt = new Date(snapshot.createdAt);
|
||||
@@ -71,9 +84,10 @@ export class EngineV2ExecutionReader {
|
||||
startedAt,
|
||||
stoppedAt: snapshot.finishedAt ? new Date(snapshot.finishedAt) : undefined,
|
||||
storedAt: 'db',
|
||||
// Step data is not mapped yet, so there is no run data to report.
|
||||
data: createEmptyRunExecutionData(),
|
||||
workflowData,
|
||||
data,
|
||||
// The same projection the v1 path reports, so the editor sees one shape.
|
||||
// The cast: the declared type overstates what either path returns.
|
||||
workflowData: toWorkflowSnapshot(workflow) as IExecutionResponse['workflowData'],
|
||||
// The data plane stores neither.
|
||||
customData: {},
|
||||
annotation: { tags: [] },
|
||||
|
||||
@@ -18,6 +18,16 @@ export type WorkflowSnapshot = Pick<
|
||||
'id' | 'name' | 'nodes' | 'connections' | 'settings' | 'nodeGroups'
|
||||
>;
|
||||
|
||||
/**
|
||||
* Narrows a workflow to what an execution read reports. Every path that serves
|
||||
* `IExecutionResponse.workflowData` must use this: the editor renders a v1 and a
|
||||
* v2 execution with the same code.
|
||||
*/
|
||||
export function toWorkflowSnapshot(workflow: WorkflowSnapshot): WorkflowSnapshot {
|
||||
const { id, name, nodes, connections, settings, nodeGroups } = workflow;
|
||||
return { id, name, nodes, connections, settings, nodeGroups };
|
||||
}
|
||||
|
||||
export type ExecutionDataPayload = BundleWorkflowSnapshot & {
|
||||
data: string;
|
||||
};
|
||||
|
||||
@@ -35,7 +35,7 @@ import {
|
||||
type BundleWorkflowSnapshot,
|
||||
type ExecutionDataPayload,
|
||||
type ExecutionRef,
|
||||
type WorkflowSnapshot,
|
||||
toWorkflowSnapshot,
|
||||
} from './execution-data/types';
|
||||
import { UnreadableRunDataError } from './execution-data/unreadable-run-data.error';
|
||||
import { sumBinaryDataBytes } from './sum-binary-data-bytes';
|
||||
@@ -98,15 +98,8 @@ export class ExecutionPersistence {
|
||||
*/
|
||||
async create(payload: CreateExecutionPayload, ctx: OperationContext = {}): Promise<string> {
|
||||
const { data: rawData, workflowData, ...rest } = payload;
|
||||
const { connections, nodes, name, settings, id, nodeGroups } = workflowData;
|
||||
const workflowSnapshot: WorkflowSnapshot = {
|
||||
connections,
|
||||
nodes,
|
||||
name,
|
||||
settings,
|
||||
id,
|
||||
nodeGroups,
|
||||
};
|
||||
const { id } = workflowData;
|
||||
const workflowSnapshot = toWorkflowSnapshot(workflowData);
|
||||
const storedAt = this.storageConfig.modeTag;
|
||||
const workflowVersionId = workflowData.versionId ?? null;
|
||||
const executionEntity = { ...rest, createdAt: new Date(), storedAt, workflowVersionId };
|
||||
@@ -811,7 +804,7 @@ export class ExecutionPersistence {
|
||||
const jsonSizeBytes = await this.trackWrite(mode, ref.workflowId, async () => {
|
||||
const bundle: ExecutionDataPayload = {
|
||||
data: stringify(data),
|
||||
workflowData: this.toWorkflowSnapshot(workflowData),
|
||||
workflowData: toWorkflowSnapshot(workflowData),
|
||||
workflowVersionId,
|
||||
};
|
||||
|
||||
@@ -851,7 +844,7 @@ export class ExecutionPersistence {
|
||||
|
||||
const bundle: ExecutionDataPayload = {
|
||||
data: serializedData,
|
||||
workflowData: workflowData ? this.toWorkflowSnapshot(workflowData) : stored.workflowData,
|
||||
workflowData: workflowData ? toWorkflowSnapshot(workflowData) : stored.workflowData,
|
||||
workflowVersionId: stored.workflowVersionId,
|
||||
};
|
||||
|
||||
@@ -994,13 +987,6 @@ export class ExecutionPersistence {
|
||||
return tx ? await this.dbStore.read(ref, tx) : await this.dbStore.read(ref);
|
||||
}
|
||||
|
||||
private toWorkflowSnapshot(
|
||||
workflowData: NonNullable<IExecutionResponse['workflowData']>,
|
||||
): WorkflowSnapshot {
|
||||
const { id, name, nodes, connections, settings, nodeGroups } = workflowData;
|
||||
return { id, name, nodes, connections, settings, nodeGroups };
|
||||
}
|
||||
|
||||
private async assembleExecution(
|
||||
entity: ExecutionEntity,
|
||||
bundle: ExecutionDataPayload,
|
||||
|
||||
@@ -189,11 +189,25 @@ describe('EngineDataPlaneClient', () => {
|
||||
expect.objectContaining({
|
||||
url: `/api/workflow-executions/${EXECUTION_ID}`,
|
||||
method: 'GET',
|
||||
qs: undefined,
|
||||
disableFollowRedirect: true,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('asks for the steps on the same request, so a read is one round trip', async () => {
|
||||
const snapshot = { id: EXECUTION_ID, workflowId: 'wf-1', steps: [] };
|
||||
respondWith(200, snapshot);
|
||||
|
||||
await expect(client.getExecution(EXECUTION_ID, { includeSteps: true })).resolves.toEqual(
|
||||
snapshot,
|
||||
);
|
||||
|
||||
expect(http.request).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ qs: { includeSteps: 'true' } }),
|
||||
);
|
||||
});
|
||||
|
||||
it('returns undefined for an execution the engine does not have', async () => {
|
||||
respondWith(404, { error: 'not_found' });
|
||||
|
||||
|
||||
@@ -83,10 +83,15 @@ export class EngineDataPlaneClient implements EngineDataPlaneProvider {
|
||||
return response.body as StartExecutionResult;
|
||||
}
|
||||
|
||||
async getExecution(id: ExecutionIdV2): Promise<ExecutionSnapshot | undefined> {
|
||||
async getExecution(
|
||||
id: ExecutionIdV2,
|
||||
options?: { includeSteps?: boolean },
|
||||
): Promise<ExecutionSnapshot | undefined> {
|
||||
const response = await this.http.request<ExecutionSnapshot | EngineErrorResponse>({
|
||||
url: `/api/workflow-executions/${encodeURIComponent(id)}`,
|
||||
method: 'GET',
|
||||
// The engine accepts only `true` or `false`.
|
||||
qs: options?.includeSteps ? { includeSteps: 'true' } : undefined,
|
||||
json: true,
|
||||
returnFullResponse: true,
|
||||
ignoreHttpStatusErrors: true,
|
||||
|
||||
@@ -56,6 +56,15 @@ describe('EngineDataPlaneProxyService', () => {
|
||||
proxy.registerProvider(provider);
|
||||
|
||||
await expect(proxy.getExecution(executionId)).resolves.toBe(snapshot);
|
||||
expect(provider.getExecution).toHaveBeenCalledWith(executionId);
|
||||
expect(provider.getExecution).toHaveBeenCalledWith(executionId, undefined);
|
||||
});
|
||||
|
||||
it('passes the read options through to the provider', async () => {
|
||||
const provider = mock<EngineDataPlaneProvider>();
|
||||
proxy.registerProvider(provider);
|
||||
|
||||
await proxy.getExecution(executionId, { includeSteps: true });
|
||||
|
||||
expect(provider.getExecution).toHaveBeenCalledWith(executionId, { includeSteps: true });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,8 +13,15 @@ import type { ExecutionIdV2 } from '@/executions/execution-id';
|
||||
export interface EngineDataPlaneProvider {
|
||||
startExecution(request: StartExecutionRequest): Promise<StartExecutionResult>;
|
||||
|
||||
/** `undefined` when the data plane holds no execution under that id. */
|
||||
getExecution(id: ExecutionIdV2): Promise<ExecutionSnapshot | undefined>;
|
||||
/**
|
||||
* `undefined` when the data plane holds no execution under that id.
|
||||
*
|
||||
* @param options.includeSteps Also report the steps, on the same round trip.
|
||||
*/
|
||||
getExecution(
|
||||
id: ExecutionIdV2,
|
||||
options?: { includeSteps?: boolean },
|
||||
): Promise<ExecutionSnapshot | undefined>;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -48,9 +55,12 @@ export class EngineDataPlaneProxyService implements EngineDataPlaneProvider {
|
||||
}
|
||||
|
||||
/** No provider means no v2 execution can exist, so this is a miss, not an error. */
|
||||
async getExecution(id: ExecutionIdV2): Promise<ExecutionSnapshot | undefined> {
|
||||
async getExecution(
|
||||
id: ExecutionIdV2,
|
||||
options?: { includeSteps?: boolean },
|
||||
): Promise<ExecutionSnapshot | undefined> {
|
||||
if (!this.provider) return undefined;
|
||||
|
||||
return await this.provider.getExecution(id);
|
||||
return await this.provider.getExecution(id, options);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,7 +8,8 @@ import {
|
||||
} from '@n8n/backend-test-utils';
|
||||
import type { User } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import type { ExecutionSnapshot } from '@n8n/engine';
|
||||
import type { ExecutionSnapshot, StepDetail } from '@n8n/engine';
|
||||
import { parse } from 'flatted';
|
||||
|
||||
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
|
||||
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
|
||||
@@ -151,15 +152,16 @@ describe('GET /executions/:id', () => {
|
||||
getExecution.mockReset();
|
||||
});
|
||||
|
||||
const snapshot = (workflowId: string): ExecutionSnapshot => ({
|
||||
const snapshot = (workflowId: string, steps?: StepDetail[]): ExecutionSnapshot => ({
|
||||
id: V2_EXECUTION_ID,
|
||||
workflowId,
|
||||
status: 'completed',
|
||||
mode: 'manual',
|
||||
graph: { nodes: [], edges: [] },
|
||||
graph: { nodes: [{ id: 'trigger-id', name: 'Trigger', type: 'trigger' }], edges: [] },
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
updatedAt: '2026-08-25T10:00:05.000Z',
|
||||
finishedAt: '2026-08-25T10:00:05.000Z',
|
||||
steps,
|
||||
});
|
||||
|
||||
test('serves a uuid id from the data plane', async () => {
|
||||
@@ -171,7 +173,7 @@ describe('GET /executions/:id', () => {
|
||||
.get(`/executions/${V2_EXECUTION_ID}`)
|
||||
.expect(200);
|
||||
|
||||
expect(getExecution).toHaveBeenCalledWith(V2_EXECUTION_ID);
|
||||
expect(getExecution).toHaveBeenCalledWith(V2_EXECUTION_ID, { includeSteps: true });
|
||||
expect(response.body.data).toMatchObject({
|
||||
id: V2_EXECUTION_ID,
|
||||
workflowId: workflow.id,
|
||||
@@ -183,6 +185,38 @@ describe('GET /executions/:id', () => {
|
||||
expect(response.body.data.workflowData.id).toBe(workflow.id);
|
||||
});
|
||||
|
||||
test('serves the step outputs as v1 run data', async () => {
|
||||
const workflow = await createWorkflow({}, owner);
|
||||
getExecution.mockResolvedValue(
|
||||
snapshot(workflow.id, [
|
||||
{
|
||||
id: 'step-1',
|
||||
nodeId: 'trigger-id',
|
||||
iteration: 0,
|
||||
status: 'completed',
|
||||
outputs: [[{ json: { hello: 'world' } }]],
|
||||
error: null,
|
||||
createdAt: '2026-08-25T10:00:00.000Z',
|
||||
updatedAt: '2026-08-25T10:00:00.250Z',
|
||||
},
|
||||
]),
|
||||
);
|
||||
|
||||
const response = await testServer
|
||||
.authAgentFor(owner)
|
||||
.get(`/executions/${V2_EXECUTION_ID}`)
|
||||
.expect(200);
|
||||
|
||||
// `data` goes out flatted, the same as a v1 execution's.
|
||||
const data = parse(response.body.data.data);
|
||||
expect(data.resultData.runData.Trigger[0]).toMatchObject({
|
||||
executionStatus: 'success',
|
||||
executionTime: 250,
|
||||
data: { main: [[{ json: { hello: 'world' } }]] },
|
||||
});
|
||||
expect(data.resultData.lastNodeExecuted).toBe('Trigger');
|
||||
});
|
||||
|
||||
test('does not serve an execution whose workflow the caller cannot read', async () => {
|
||||
const workflow = await createWorkflow({}, owner);
|
||||
// Give the member a workflow, so the request reaches the reader.
|
||||
@@ -207,7 +241,7 @@ describe('GET /executions/:id', () => {
|
||||
.expect(200);
|
||||
const v1 = await testServer.authAgentFor(owner).get('/executions/999999').expect(200);
|
||||
|
||||
expect(getExecution).toHaveBeenCalledWith(V2_EXECUTION_ID);
|
||||
expect(getExecution).toHaveBeenCalledWith(V2_EXECUTION_ID, { includeSteps: true });
|
||||
// The id was understood; there is just nothing behind it.
|
||||
expect(v2.body).toEqual(v1.body);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user